Fanout-Muster

Werfen wir einen kurzen Blick auf das Fanout-Muster in Go. Im Allgemeinen wird Fanout verwendet, um mehrere Aufgaben gleichzeitig auszuführen.

Angenommen, Sie haben eine Datenpipeline und möchten die einzelnen Elemente verarbeiten. Wir können Go-Routinen und -Kanäle verwenden, um die Elemente aufzuteilen, sobald wir sie erhalten, und dann die einzelnen Elemente verarbeiten (z. B. in dB eingeben).

Es ist ein einfach zu implementierendes Muster; Sie müssen die Kanäle jedoch verwalten, um einen Deadlock zu verhindern.

// produce is simulating our single input as a channel
func produce(id int) chan int {
    ch := make(chan int)
    go func() {
        for i := 0; i < 10; i++ {
            ch <- rand.Intn(20)
        }
        fmt.Printf("producer %d done\n", id)
        close(ch) // this is important!!!
    }()
    return ch
}

func worker(id int, jobs chan int, wg *sync.WaitGroup) {
    for value := range jobs {
        odd := "even"
        if (value & 1) == 1 {
            odd = "odd"
        }
        fmt.Printf("worker: %d, got %d is %s\n", id, value, odd)
    }
    wg.Done()
}

func main() {
    inputCh := produce(1)

    numWorkers := 3
    jobs := make(chan int)

    // split input into individual jobs
    go func() {
        for value := range inputCh {
            jobs <- value
        }
        close(jobs) // this is important!!!
    }()

    // fan-out
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go worker(i, jobs, &wg)
    }
    wg.Wait()

    fmt.Println("done")
}

Die Grundidee hierbei ist, dass es eine Abfolge von Daten gibt, die von einer festen Anzahl von Arbeitern bearbeitet werden müssen.

Für die Eingabe erstellen wir eine Folge von Zufallszahlen und platzieren diese in einem Kanal. Wir verlagern sie in einen anderen Kanal, aus dem die Arbeiter ihre „Arbeitsplätze“ ziehen.

In diesem Beispiel ist es nicht unbedingt erforderlich, die Eingabe in den Jobs-Kanal zu verschieben. Wir könnten die Arbeiter genauso gut aus dem Eingabekanal ziehen lassen; Dies geschieht hier nur aus Gründen der Klarheit.

Wir senden dann die festgelegte Anzahl an Workern als Goroutinen an den Start. Jeder Arbeiter ruft den Auftragskanal ab, bis keine Daten mehr verarbeitet werden müssen. Zu diesem Zeitpunkt signalisiert er einer WaitGroup, dass er fertig ist.

Der Hauptthread verwendet eine WaitGroup, um sicherzustellen, dass er erst abgeschlossen wird, wenn alle Arbeiter fertig sind, d. h. alle Jobs verarbeitet wurden.

Ein wichtiger Punkt ist zu erwähnen, dass dieses Muster keine Garantien für die Reihenfolge der Verarbeitung der Eingabesequenz bietet. Dies kann unter vielen Umständen in Ordnung sein. Die Eingabesequenz besteht beispielsweise aus Datensätzen, die ihren eigenen Zeitstempel enthalten, und das Ziel besteht darin, die Datensätze in dB zu speichern. Fan-Out wäre in diesem Fall akzeptabel.

Eine letzte Anmerkung: Sie werden einige Kommentare zum Schließen der Kanäle sehen, sobald alle Daten in der Sequenz gesendet wurden. Das ist entscheidend. Der Bereichsoperator, der die Kanäle abruft, geht in den Ruhezustand, sobald keine Daten mehr vorhanden sind. Sie können dies überprüfen, indem Sie eine der close()-Anweisungen auskommentieren, die einen Deadlock-Zustand verursachen. Goroutinen und Kanäle sind sehr mächtig, aber man muss sie mit Bedacht einsetzen.

Was würdest du anders machen? Wie können wir dieses Beispiel verbessern? Hinterlassen Sie unten Ihre Kommentare.

Danke!

Den Code für diesen Beitrag und alle Beiträge dieser Reihe finden Sie hier

Das obige ist der detaillierte Inhalt vonFanout-Muster. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!