Go Singleflight Melts in Ihrem Code, nicht in Ihrer Datenbank

Der Originalartikel ist im VictoriaMetrics-Blog veröffentlicht: https://victoriametrics.com/blog/go-singleflight/

Dieser Beitrag ist Teil einer Serie über den Umgang mit Parallelität in Go:

• Go sync.Mutex: Normal- und Hungermodus
• Gehen Sie zu sync.WaitGroup und dem Ausrichtungsproblem
• Go sync.Pool und die Mechanismen dahinter
• Gehen Sie zu sync.Cond, dem am meisten übersehenen Synchronisierungsmechanismus
• Go sync.Map: Das richtige Tool für den richtigen Job
• Go Sync.Once ist einfach... Stimmt das wirklich?
• Go Singleflight schmilzt in Ihrem Code, nicht in Ihrer Datenbank (wir sind hier)

Wenn also mehrere Anfragen gleichzeitig eingehen, die nach den gleichen Daten fragen, ist das Standardverhalten, dass jede dieser Anfragen einzeln an die Datenbank geht, um die gleichen Informationen zu erhalten . Das bedeutet, dass Sie am Ende dieselbe Abfrage mehrmals ausführen würden, was, seien wir ehrlich, einfach ineffizient ist.

Am Ende wird dadurch Ihre Datenbank unnötig belastet, was alles verlangsamen könnte, aber es gibt einen Weg, dies zu umgehen.

Die Idee ist, dass nur die erste Anfrage tatsächlich an die Datenbank geht. Der Rest der Anfragen wartet darauf, dass die erste abgeschlossen wird. Sobald die Daten von der ersten Anfrage zurückkommen, erhalten die anderen einfach das gleiche Ergebnis – es sind keine zusätzlichen Abfragen erforderlich.

Jetzt haben Sie also eine ziemlich gute Vorstellung davon, worum es in diesem Beitrag geht, oder?

Einzelflug

Das Singleflight-Paket in Go wurde speziell für genau das entwickelt, worüber wir gerade gesprochen haben. Und nur als Hinweis: Es ist nicht Teil der Standardbibliothek, wird aber vom Go-Team gepflegt und entwickelt.

Singleflight stellt sicher, dass nur eine dieser Goroutinen den Vorgang tatsächlich ausführt, z. B. das Abrufen der Daten aus der Datenbank. Es erlaubt zu jedem Zeitpunkt nur einen „in-flight“ (laufenden) Vorgang für dasselbe Datenelement (bekannt als „Schlüssel“).

Wenn also andere Goroutinen nach denselben Daten (dem gleichen Schlüssel) fragen, während dieser Vorgang noch läuft, warten sie einfach. Wenn dann der erste Vorgang abgeschlossen ist, erhalten alle anderen das gleiche Ergebnis, ohne dass der Vorgang erneut ausgeführt werden muss.

Okay, genug geredet, lass uns in eine kurze Demo eintauchen, um zu sehen, wie Singleflight in Aktion funktioniert:

var callCount atomic.Int32
var wg sync.WaitGroup

// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
    callCount.Add(1)
    time.Sleep(100 * time.Millisecond)
    return rand.Intn(100), nil
}

// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
    defer wg.Done()

    time.Sleep(time.Duration(id) * 40 * time.Millisecond)
    v, err, shared := g.Do("key-fetch-data", fetchData)
    if err != nil {
        return err
    }

    fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
    return nil
}

func main() {
    var g singleflight.Group

    // 5 goroutines to fetch the same data
    const numGoroutines = 5
    wg.Add(numGoroutines)

    for i := 0; i < numGoroutines; i++ {
        go fetchDataWrapper(&g, i)
    }

    wg.Wait()
    fmt.Printf("Function was called %d times\n", callCount.Load())
}

// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times

Was ist hier los:

Wir simulieren eine Situation, in der 5 Goroutinen versuchen, fast gleichzeitig im Abstand von 60 ms die gleichen Daten abzurufen. Der Einfachheit halber verwenden wir Zufallszahlen, um aus einer Datenbank abgerufene Daten nachzuahmen.

Mit singleflight.Group stellen wir sicher, dass nur die erste Goroutine tatsächlich fetchData() ausführt und der Rest auf das Ergebnis wartet.

Die Zeile v, err, shared := g.Do("key-fetch-data", fetchData) weist einen eindeutigen Schlüssel („key-fetch-data“) zu, um diese Anfragen zu verfolgen. Wenn also eine andere Goroutine nach demselben Schlüssel fragt, während die erste noch die Daten abruft, wartet sie auf das Ergebnis, anstatt einen neuen Aufruf zu starten.

Sobald der erste Aufruf beendet ist, erhalten alle wartenden Goroutinen das gleiche Ergebnis, wie wir in der Ausgabe sehen können. Obwohl wir 5 Goroutinen hatten, die nach den Daten fragten, wurde fetchData nur zweimal ausgeführt, was einen enormen Schub darstellt.

Das Shared-Flag bestätigt, dass das Ergebnis über mehrere Goroutinen hinweg wiederverwendet wurde.

"Aber warum ist die Shared-Flagge für die erste Goroutine wahr? Ich dachte, nur die Wartenden hätten geteilt == true?"

Ja, das fühlt sich vielleicht etwas kontraintuitiv an, wenn Sie denken, dass nur die wartenden Goroutinen geteilt haben sollten == wahr.

Die Sache ist, dass die gemeinsam genutzte Variable in g.Do Ihnen sagt, ob das Ergebnis von mehreren Aufrufern geteilt wurde. Im Grunde heißt es: „Hey, dieses Ergebnis wurde von mehr als einem Anrufer verwendet.“ Es geht nicht darum, wer die Funktion ausgeführt hat, es ist nur ein Signal, dass das Ergebnis in mehreren Goroutinen wiederverwendet wurde.

"Ich habe einen Cache, warum brauche ich einen Singleflight?"

Die kurze Antwort lautet: Caches und Singleflight lösen unterschiedliche Probleme und funktionieren tatsächlich sehr gut zusammen.

In einem Setup mit einem externen Cache (wie Redis oder Memcached) fügt Singleflight eine zusätzliche Schutzschicht hinzu, nicht nur für Ihre Datenbank, sondern auch für den Cache selbst.

Darüber hinaus trägt Singleflight zum Schutz vor einem Cache-Miss-Sturm (manchmal auch „Cache-Stampede“ genannt) bei.

Normalerweise ist es ein Cache-Treffer, wenn eine Anfrage nach Daten fragt und sich die Daten im Cache befinden. Wenn sich die Daten nicht im Cache befinden, handelt es sich um einen Cache-Fehler. Angenommen, 10.000 Anfragen treffen auf einmal im System ein, bevor der Cache neu aufgebaut wird, dann könnte die Datenbank plötzlich mit 10.000 identischen Abfragen gleichzeitig überlastet werden.

Während dieser Spitzenzeit stellt Singleflight sicher, dass nur eine dieser 10.000 Anfragen tatsächlich die Datenbank erreicht.

Aber später, im Abschnitt zur internen Implementierung, werden wir sehen, dass Singleflight eine globale Sperre verwendet, um die Karte von In-Flight-Aufrufen zu schützen, die zu einem einzigen Streitpunkt für jede Goroutine werden können. Dies kann die Geschwindigkeit verlangsamen, insbesondere wenn Sie mit hoher Parallelität arbeiten.

Das folgende Modell funktioniert möglicherweise besser für Maschinen mit mehreren CPUs:

In diesem Setup verwenden wir Singleflight nur, wenn ein Cache-Fehler auftritt.

Einzelflugbetrieb

Um Singleflight zu verwenden, erstellen Sie zunächst ein Gruppenobjekt, das die Kernstruktur darstellt, die laufende Funktionsaufrufe verfolgt, die mit bestimmten Tasten verknüpft sind.

Es gibt zwei Schlüsselmethoden, die helfen, doppelte Anrufe zu verhindern:

• group.Do(key, func): Führt Ihre Funktion aus und unterdrückt gleichzeitig doppelte Anforderungen. Wenn Sie Do aufrufen, übergeben Sie einen Schlüssel und eine Funktion. Wenn für diesen Schlüssel keine andere Ausführung stattfindet, wird die Funktion ausgeführt. Wenn für denselben Schlüssel bereits eine Ausführung ausgeführt wird, wird Ihr Aufruf blockiert, bis der erste abgeschlossen ist und das gleiche Ergebnis zurückgibt.
• group.DoChan(key, func): Ähnlich wie group.Do, aber anstatt zu blockieren, gibt es Ihnen einen Kanal (<-chan-Ergebnis). Sie erhalten das Ergebnis, sobald es fertig ist. Dies ist nützlich, wenn Sie das Ergebnis lieber asynchron verarbeiten oder über mehrere Kanäle auswählen möchten.

Wir haben bereits in der Demo gesehen, wie man g.DoChan() verwendet. Sehen wir uns nun an, wie man g.DoChan() mit einer modifizierten Wrapper-Funktion verwendet:

var callCount atomic.Int32
var wg sync.WaitGroup

// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
    callCount.Add(1)
    time.Sleep(100 * time.Millisecond)
    return rand.Intn(100), nil
}

// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
    defer wg.Done()

    time.Sleep(time.Duration(id) * 40 * time.Millisecond)
    v, err, shared := g.Do("key-fetch-data", fetchData)
    if err != nil {
        return err
    }

    fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
    return nil
}

func main() {
    var g singleflight.Group

    // 5 goroutines to fetch the same data
    const numGoroutines = 5
    wg.Add(numGoroutines)

    for i := 0; i < numGoroutines; i++ {
        go fetchDataWrapper(&g, i)
    }

    wg.Wait()
    fmt.Printf("Function was called %d times\n", callCount.Load())
}

// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times

// Wrap the fetchData function with singleflight using DoChan
func fetchDataWrapper(g *singleflight.Group, id int) error {
    defer wg.Done()

    ch := g.DoChan("key-fetch-data", fetchData)

    res := <-ch
    if res.Err != nil {
        return res.Err
    }

    fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, res.Val, res.Shared)
    return nil
}

Um ehrlich zu sein, ändert die Verwendung von DoChan() hier im Vergleich zu Do() nicht viel, da wir immer noch auf das Ergebnis mit einer Kanalempfangsoperation (<-ch) warten, die im Grunde dasselbe blockiert Weg.

DoChan() glänzt, wenn Sie einen Vorgang starten und andere Dinge erledigen möchten, ohne die Goroutine zu blockieren. Mit den Kanälen:
könnten Sie beispielsweise Zeitüberschreitungen oder Absagen sauberer handhaben

package singleflight

type Result struct {
    Val    interface{}
    Err    error
    Shared bool
}

Dieses Beispiel wirft auch einige Probleme auf, auf die Sie in realen Szenarien stoßen könnten:

• Die erste Goroutine kann aufgrund langsamer Netzwerkantworten, nicht reagierender Datenbanken usw. viel länger dauern als erwartet. In diesem Fall bleiben alle anderen wartenden Goroutinen länger hängen, als Ihnen lieb ist. Eine Auszeit kann hier Abhilfe schaffen, aber alle neuen Anfragen bleiben am Ende immer noch hinter der ersten zurück.
• Die von Ihnen abgerufenen Daten können sich häufig ändern, sodass das Ergebnis möglicherweise veraltet ist, wenn die erste Anfrage abgeschlossen ist. Das heißt, wir brauchen eine Möglichkeit, den Schlüssel ungültig zu machen und eine neue Ausführung auszulösen.

Ja, Singleflight bietet eine Möglichkeit, Situationen wie diese mit der Methode „group.Forget(key)“ zu bewältigen, mit der Sie eine laufende Ausführung verwerfen können.

Die Forget()-Methode entfernt einen Schlüssel aus der internen Karte, der die laufenden Funktionsaufrufe verfolgt. Es ist so, als würde man den Schlüssel „ungültig machen“. Wenn Sie also g.Do() mit diesem Schlüssel erneut aufrufen, wird die Funktion so ausgeführt, als wäre es eine neue Anfrage, anstatt auf den Abschluss der vorherigen Ausführung zu warten.

Aktualisieren wir unser Beispiel, um Forget() zu verwenden und sehen, wie oft die Funktion tatsächlich aufgerufen wird:

var callCount atomic.Int32
var wg sync.WaitGroup

// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
    callCount.Add(1)
    time.Sleep(100 * time.Millisecond)
    return rand.Intn(100), nil
}

// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
    defer wg.Done()

    time.Sleep(time.Duration(id) * 40 * time.Millisecond)
    v, err, shared := g.Do("key-fetch-data", fetchData)
    if err != nil {
        return err
    }

    fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
    return nil
}

func main() {
    var g singleflight.Group

    // 5 goroutines to fetch the same data
    const numGoroutines = 5
    wg.Add(numGoroutines)

    for i := 0; i < numGoroutines; i++ {
        go fetchDataWrapper(&g, i)
    }

    wg.Wait()
    fmt.Printf("Function was called %d times\n", callCount.Load())
}

// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times

Goroutine 0 und Goroutine 1 rufen beide Do() mit demselben Schlüssel („key-fetch-data“) auf, und ihre Anforderungen werden in einer Ausführung zusammengefasst und das Ergebnis wird zwischen den beiden Goroutinen geteilt.

Goroutine 2 hingegen ruft Forget() auf, bevor Do() ausgeführt wird. Dadurch werden alle vorherigen Ergebnisse gelöscht, die mit „key-fetch-data“ verknüpft sind, sodass eine neue Ausführung der Funktion ausgelöst wird.

Zusammenfassend lässt sich sagen, dass Singleflight zwar nützlich ist, aber dennoch einige Randfälle haben kann, zum Beispiel:

• Wenn die erste Goroutine zu lange blockiert wird, bleiben auch alle anderen, die darauf warten, hängen. In solchen Fällen kann die Verwendung eines Timeout-Kontexts oder einer Select-Anweisung mit Timeout eine bessere Option sein.
• Wenn die erste Anfrage einen Fehler oder eine Panik zurückgibt, wird derselbe Fehler oder die gleiche Panik auf alle anderen Goroutinen übertragen, die auf das Ergebnis warten.

Wenn Ihnen alle von uns besprochenen Probleme aufgefallen sind, tauchen wir im nächsten Abschnitt ein, um zu besprechen, wie Singleflight tatsächlich unter der Haube funktioniert.

So funktioniert Singleflight

Durch die Verwendung von Singleflight haben Sie möglicherweise bereits eine grundlegende Vorstellung davon, wie es intern funktioniert. Die gesamte Implementierung von Singleflight umfasst nur etwa 150 Codezeilen.

Grundsätzlich erhält jeder eindeutige Schlüssel eine Struktur, die seine Ausführung verwaltet. Wenn eine Goroutine Do() aufruft und feststellt, dass der Schlüssel bereits existiert, wird dieser Aufruf blockiert, bis die erste Ausführung abgeschlossen ist, und hier ist die Struktur:

// Wrap the fetchData function with singleflight using DoChan
func fetchDataWrapper(g *singleflight.Group, id int) error {
    defer wg.Done()

    ch := g.DoChan("key-fetch-data", fetchData)

    res := <-ch
    if res.Err != nil {
        return res.Err
    }

    fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, res.Val, res.Shared)
    return nil
}

Hier werden zwei Synchronisierungsprimitive verwendet:

• Gruppenmutex (g.mu): Dieser Mutex schützt die gesamte Schlüsselzuordnung, nicht ein Schloss pro Schlüssel, er stellt sicher, dass das Hinzufügen oder Entfernen von Schlüsseln threadsicher ist.
• WaitGroup (g.call.wg): Die WaitGroup wird verwendet, um darauf zu warten, dass die erste Goroutine, die einer bestimmten Taste zugeordnet ist, ihre Arbeit beendet.

Wir konzentrieren uns hier auf die Methode group.Do(), da die andere Methode, group.DoChan(), auf ähnliche Weise funktioniert. Die Methode group.Forget() ist ebenfalls einfach, da sie lediglich den Schlüssel aus der Karte entfernt.

Wenn Sie group.Do() aufrufen, sperrt es zunächst die gesamte Aufrufliste (g.mu).

„Ist das nicht schlecht für die Leistung?“

Ja, es ist möglicherweise nicht in jedem Fall ideal für die Leistung (es ist immer gut, zuerst einen Benchmark durchzuführen), da Singleflight die gesamten Schlüssel sperrt. Wenn Sie eine bessere Leistung anstreben oder in großem Umfang arbeiten, ist das Sharding oder Verteilen der Schlüssel ein guter Ansatz. Anstatt nur eine Singleflight-Gruppe zu verwenden, können Sie die Last auf mehrere Gruppen verteilen, ähnlich wie Sie stattdessen „Multiflight“ verwenden

Als Referenz sehen Sie sich dieses Repo an: shardedsingleflight.

Sobald die Sperre aktiviert ist, prüft die Gruppe anhand der internen Karte (g.m), ob für den angegebenen Schlüssel bereits ein laufender oder abgeschlossener Anruf vorliegt. Diese Karte verfolgt alle laufenden oder abgeschlossenen Arbeiten, wobei die Tasten den entsprechenden Aufgaben zugeordnet sind.

Wenn der Schlüssel gefunden wird (eine andere Goroutine führt die Aufgabe bereits aus), erhöhen wir einfach einen Zähler (c.dups), um doppelte Anforderungen zu verfolgen, anstatt einen neuen Aufruf zu starten. Anschließend gibt die Goroutine die Sperre frei und wartet auf den Abschluss der ursprünglichen Aufgabe, indem sie call.wg.Wait() für die zugehörige WaitGroup aufruft.

Wenn die ursprüngliche Aufgabe erledigt ist, greift diese Goroutine auf das Ergebnis und vermeidet die erneute Ausführung der Aufgabe.

var callCount atomic.Int32
var wg sync.WaitGroup

// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
    callCount.Add(1)
    time.Sleep(100 * time.Millisecond)
    return rand.Intn(100), nil
}

// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
    defer wg.Done()

    time.Sleep(time.Duration(id) * 40 * time.Millisecond)
    v, err, shared := g.Do("key-fetch-data", fetchData)
    if err != nil {
        return err
    }

    fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
    return nil
}

func main() {
    var g singleflight.Group

    // 5 goroutines to fetch the same data
    const numGoroutines = 5
    wg.Add(numGoroutines)

    for i := 0; i < numGoroutines; i++ {
        go fetchDataWrapper(&g, i)
    }

    wg.Wait()
    fmt.Printf("Function was called %d times\n", callCount.Load())
}

// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times

Wenn keine andere Goroutine an diesem Schlüssel arbeitet, übernimmt die aktuelle Goroutine die Verantwortung für die Ausführung der Aufgabe.

An diesem Punkt erstellen wir ein neues Anrufobjekt, fügen es der Karte hinzu und initialisieren seine WaitGroup. Dann entsperren wir den Mutex und führen die Aufgabe selbst über eine Hilfsmethode g.doCall(c, key, fn) aus. Wenn die Aufgabe abgeschlossen ist, werden alle wartenden Goroutinen durch den Aufruf wg.Wait() entsperrt.

Hier gibt es nichts allzu Ungewöhnliches, außer wie wir mit Fehlern umgehen, es gibt drei mögliche Szenarien:

• Wenn die Funktion in Panik gerät, fangen wir sie ab, verpacken sie in einen panicError und lösen die Panik aus.
• Wenn die Funktion ein errGoexit zurückgibt, rufen wir runtime.Goexit() auf, um die Goroutine ordnungsgemäß zu beenden.
• Wenn es sich nur um einen normalen Fehler handelt, setzen wir diesen Fehler beim Anruf.

Hier wird es mit der Hilfsmethode g.doCall() etwas cleverer.

"Warte, was ist runtime.Goexit()?"

Bevor wir in den Code eintauchen, möchte ich kurz erklären, dass runtime.Goexit() verwendet wird, um die Ausführung einer Goroutine zu stoppen.

Wenn eine Goroutine Goexit() aufruft, stoppt sie und alle verzögerten Funktionen werden weiterhin wie gewohnt in der Last-In-First-Out (LIFO)-Reihenfolge ausgeführt. Es ähnelt einer Panik, es gibt jedoch ein paar Unterschiede:

• Es löst keine Panik aus, Sie können es also nicht mit „recover()“ abfangen.
• Nur ​​die Goroutine, die Goexit() aufruft, wird beendet und alle anderen Goroutinen laufen einwandfrei weiter.

Nun, hier ist eine interessante Eigenart (die nicht direkt mit unserem Thema zusammenhängt, aber erwähnenswert ist). Wenn Sie runtime.Goexit() in der Haupt-Goroutine aufrufen (wie in main()), sehen Sie sich Folgendes an:

var callCount atomic.Int32
var wg sync.WaitGroup

// Simulate a function that fetches data from a database
func fetchData() (interface{}, error) {
    callCount.Add(1)
    time.Sleep(100 * time.Millisecond)
    return rand.Intn(100), nil
}

// Wrap the fetchData function with singleflight
func fetchDataWrapper(g *singleflight.Group, id int) error {
    defer wg.Done()

    time.Sleep(time.Duration(id) * 40 * time.Millisecond)
    v, err, shared := g.Do("key-fetch-data", fetchData)
    if err != nil {
        return err
    }

    fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, v, shared)
    return nil
}

func main() {
    var g singleflight.Group

    // 5 goroutines to fetch the same data
    const numGoroutines = 5
    wg.Add(numGoroutines)

    for i := 0; i < numGoroutines; i++ {
        go fetchDataWrapper(&g, i)
    }

    wg.Wait()
    fmt.Printf("Function was called %d times\n", callCount.Load())
}

// Output:
// Goroutine 0: result: 90, shared: true
// Goroutine 2: result: 90, shared: true
// Goroutine 1: result: 90, shared: true
// Goroutine 3: result: 13, shared: true
// Goroutine 4: result: 13, shared: true
// Function was called 2 times

Was passiert, ist, dass Goexit() die Haupt-Goroutine beendet, aber wenn noch andere Goroutinen laufen, läuft das Programm weiter, weil die Go-Laufzeit am Leben bleibt, solange mindestens eine Goroutine aktiv ist. Sobald jedoch keine Goroutinen mehr vorhanden sind, stürzt es mit der Fehlermeldung „Keine Goroutinen“ ab, eine Art lustiger kleiner Eckfall.

Nun zurück zu unserem Code: Wenn runtime.Goexit() nur die aktuelle Goroutine beendet und nicht von restart() abgefangen werden kann, wie erkennen wir dann, ob sie aufgerufen wurde?

Der Schlüssel liegt in der Tatsache, dass beim Aufruf von runtime.Goexit() jeglicher Code danach nicht ausgeführt wird.

// Wrap the fetchData function with singleflight using DoChan
func fetchDataWrapper(g *singleflight.Group, id int) error {
    defer wg.Done()

    ch := g.DoChan("key-fetch-data", fetchData)

    res := <-ch
    if res.Err != nil {
        return res.Err
    }

    fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, res.Val, res.Shared)
    return nil
}

Im obigen Fall wird die Zeile normalReturn = true nach dem Aufruf von runtime.Goexit() nie ausgeführt. So können wir innerhalb der Verzögerung prüfen, ob normalReturn immer noch false ist, um zu erkennen, dass eine spezielle Methode aufgerufen wurde.

Der nächste Schritt besteht darin, herauszufinden, ob die Aufgabe in Panik gerät oder nicht. Dafür verwenden wir „recover()“ als normale Rückgabe, obwohl der eigentliche Code in singleflight etwas subtiler ist:

package singleflight

type Result struct {
    Val    interface{}
    Err    error
    Shared bool
}

Anstatt „recovered = true“ direkt im Recovery-Block zu setzen, wird dieser Code etwas ausgefallener, indem er „recovered“ nach dem „recover()“-Block als letzte Zeile einstellt.

Warum funktioniert das?

Wenn runtime.Goexit() aufgerufen wird, beendet es die gesamte Goroutine, genau wie ein panic(). Wenn jedoch eine Panic()-Funktion wiederhergestellt wird, wird nur die Funktionskette zwischen Panic() und Recover() beendet, nicht die gesamte Goroutine.

Aus diesem Grund wird „recovered = true“ außerhalb des Defer-Befehls gesetzt, der „recover()“ enthält, und nur in zwei Fällen ausgeführt: wenn die Funktion normal abgeschlossen wird oder wenn eine Panik wiederhergestellt wird, aber nicht, wenn runtime.Goexit() aufgerufen wird.

In Zukunft werden wir besprechen, wie mit jedem Fall umgegangen wird.

func fetchDataWrapperWithTimeout(g *singleflight.Group, id int) error {
    defer wg.Done()

    ch := g.DoChan("key-fetch-data", fetchData)
    select {
    case res := <-ch:
        if res.Err != nil {
            return res.Err
        }
        fmt.Printf("Goroutine %d: result: %v, shared: %v\n", id, res.Val, res.Shared)
    case <-time.After(50 * time.Millisecond):
        return fmt.Errorf("timeout waiting for result")
    }

  return nil
}

Wenn die Aufgabe während der Ausführung in Panik gerät, wird die Panik abgefangen und in c.err als panicError gespeichert, der sowohl den Panikwert als auch den Stack-Trace enthält. singleflight fängt die Panik auf, um elegant aufzuräumen, aber es schluckt sie nicht herunter, sondern wirft die Panik erneut aus, nachdem es seinen Zustand verarbeitet hat.

Das bedeutet, dass die Panik in der Goroutine auftritt, die die Aufgabe ausführt (die erste, die den Vorgang startet), und dass alle anderen Goroutinen, die auf das Ergebnis warten, ebenfalls in Panik geraten.

Da diese Panik im Code des Entwicklers auftritt, liegt es an uns, richtig damit umzugehen.

Nun müssen wir noch einen Sonderfall berücksichtigen: wenn andere Goroutinen die Methode group.DoChan() verwenden und über einen Kanal auf ein Ergebnis warten. In diesem Fall kann Singleflight in diesen Goroutinen nicht in Panik geraten. Stattdessen kommt es zu einer sogenannten nicht behebbaren Panik (go panic(e)), die zum Absturz unserer Anwendung führt.

Wenn die Aufgabe schließlich runtime.Goexit() aufruft, besteht keine Notwendigkeit, weitere Maßnahmen zu ergreifen, da die Goroutine bereits dabei ist, herunterzufahren, und wir lassen das einfach geschehen, ohne einzugreifen.

Und das ist so ziemlich alles, nichts allzu Kompliziertes außer den Sonderfällen, die wir besprochen haben.

Bleiben Sie in Verbindung

Hallo, ich bin Phuong Le, Softwareentwickler bei VictoriaMetrics. Der obige Schreibstil konzentriert sich auf Klarheit und Einfachheit und erklärt Konzepte auf eine leicht verständliche Weise, auch wenn sie nicht immer perfekt mit akademischer Präzision übereinstimmt.

Wenn Sie etwas entdecken, das veraltet ist, oder wenn Sie Fragen haben, zögern Sie nicht, uns zu kontaktieren. Du kannst mir eine DM an X(@func25) schicken.

Einige andere Beiträge, die Sie interessieren könnten:

• Go I/O-Leser, -Schreiber und Data in Motion.
• Wie Go-Arrays funktionieren und wie es mit For-Range knifflig wird
• Slices in Go: Groß werden oder nach Hause gehen
• Go Maps erklärt: Wie Schlüssel-Wert-Paare tatsächlich gespeichert werden
• Golang Defer: Von Basic zu Traps
• Vendoring oder Go-Mod-Vendor: Was ist das?

Wer wir sind

Wenn Sie Ihre Dienste überwachen, Metriken verfolgen und sehen möchten, wie alles funktioniert, sollten Sie sich VictoriaMetrics ansehen. Es ist eine schnelle, Open-Source und kostensparende Möglichkeit, Ihre Infrastruktur im Auge zu behalten.

Und wir sind Gophers, Enthusiasten, die es lieben, zu forschen, zu experimentieren und Wissen über Go und sein Ökosystem zu teilen.

Das obige ist der detaillierte Inhalt vonGo Singleflight Melts in Ihrem Code, nicht in Ihrer Datenbank. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!