Fan-Out/Fan-In in Go: Muster für parallele Datenverarbeitung

Fan-Out/Fan-In-Muster in Go: Flexible parallele Datenverarbeitung

Das Fan-Out/Fan-In-Muster bietet in Go eine starke Methode, um Aufgaben auf Goroutines zu verteilen und die Ergebnisse anschließend zu sammeln. Im Gegensatz zum Worker-Pool-Muster ermöglicht es durch individuelle Verarbeitungspipelines pro Worker eine höhere Flexibilität, was besonders bei komplexen Daten-Transformationsketten nützlich ist.

Grundprinzipien des Musters

Das Muster gliedert sich in zwei zentrale Phasen. In der Fan-Out-Phase werden Eingabedaten auf mehrere Worker (Goroutines) verteilt. Jeder Worker erhält seine eigene Aufgabe, verarbeitet sie unabhängig und sendet das Ergebnis über einen dedizierten Kanal. Die Fan-In-Phase fasst dann die Ergebnisse aus allen Worker-Kanälen in einem einzigen Ausgabekanal zusammen, der für die nachgelagerte Verarbeitung genutzt wird.

Der Hauptunterschied zum Worker Pool liegt in der Kanalarchitektur:

• Worker Pool: Alle Worker lesen typischerweise Aufgaben aus einem gemeinsamen Kanal und schreiben Ergebnisse in einen weiteren gemeinsamen Kanal.
• Fan-Out/Fan-In: Jeder Worker hat seinen eigenen Ausgabekanal, was einen zusätzlichen Merge-Schritt erfordert.

Dieser Ansatz verursacht durch die zusätzlichen Goroutines und Kanäle für das Merging etwas Overhead, entfaltet aber sein volles Potenzial in Pipeline-Architekturen, in denen spätere Stufen separate Datenströme benötigen.

Praktische Umsetzung

Gehen wir eine praxisnahe Implementierung durch, die bewusst vom klassischen Worker Pool abweicht, um die Kernmerkmale hervorzuheben.

Worker-Funktion

In diesem Setup ist der Worker eine Funktion, die eine Goroutine startet und einen Ergebniskanal zurückgibt. Das unterscheidet sich vom Übergaben eines Kanals als Parameter.

func worker(x int, workerID int) chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        fmt.Printf("worker %d, job %d\n", workerID, x)
        out <- x * x
    }()
    return out
}

Wichtige Implementierungsdetails:

• Die Goroutine schließt ihren out-Kanal mit defer nach dem Senden des Ergebnisses.
• Für die Demo ist die Arbeit trivial – das Quadrieren einer Zahl.
• Im Produktionscode müssten Sie das verfeinern: Lebenszyklus mit Context für saubere Abschlüsse und Timeouts managen.

Fan-Out-Phase (Verteilung)

Der Distributor (Produzent) startet pro Eingabedaten-Element einen Worker. Die von den Workern zurückgegebenen Kanäle werden in einem Slice für das spätere Merging gespeichert.

for i, x := range data {
    chnls = append(chnls, worker(x, i+1))
}

Schlüsselmerkmal: Die Anzahl der Worker skaliert dynamisch mit der Aufgabenanzahl – nicht zwingend erforderlich, zeigt aber die Flexibilität des Musters. In realen Anwendungen begrenzen Sie meist die parallelen Worker.

Fan-In-Phase und Merge-Funktion

Die merge-Funktion übernimmt die entscheidende Aufgabe, Daten aus mehreren Kanälen in einen zu bündeln. Sie:

• Erzeugt einen einzelnen ungepufferten Ausgabekanal out.
• Startet pro Eingabekanal eine Goroutine, die Daten liest und an out weiterleitet.
• Wartet in einer separaten Goroutine mit sync.WaitGroup, bis alle Leser-Goroutines fertig sind, und schließt dann out.

func merge(chnls ...chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range chnls {
        wg.Add(1)
        go func(c chan int) {
            defer wg.Done()
            for x := range c {
                out <- x
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

Technischer Hinweis: Das Warten (wg.Wait()) läuft in einer separaten Goroutine. Im Hauptfluss könnte es zu einem Deadlock kommen, falls der Consumer noch nicht aus out liest und Schreibvorgänge blockiert werden.

Consumer

Der Consumer liest Ergebnisse aus dem einzelnen gemergten Kanal. Die for range-Schleife endet automatisch, wenn merge den out-Kanal schließt.

for x := range out {
    results = append(results, x)
}

Wichtige Erkenntnisse

• Architektonische Flexibilität: Individuelle Pipelines pro Worker erleichtern das Einfügen zusätzlicher Datenverarbeitungsstufen.
• Dynamisches Skalieren: Theoretisch ein Worker pro Aufgabe möglich, in der Praxis Konkurrenz steuern.
• Overhead: Das Muster verbraucht mehr Ressourcen (Kanäle, Goroutines) für das Merging als einfachere Ansätze.
• Lebenszyklus-Management: Diese Demo verzichtet auf Contexts – für Beispiele ok, in der Produktion essenziell für Abbrüche und Timeouts.
• Code-Lesbarkeit: Ungepufferte Kanäle und for range-Schleifen halten den Code knapp und klar.

Produktionsempfehlungen

Dieser Democode dient dem Lernen. Für den realen Einsatz diese Verbesserungen vornehmen:

• Context hinzufügen (context.Context): An Worker und merge weitergeben für Abbrüche, Fristen und Timeouts.
• Konkurrenz begrenzen: Unbegrenzte Goroutines (eine pro Aufgabe) können Ressourcen erschöpfen. Semaphore oder fester Worker-Pool im Fan-Out nutzen.
• Fehlerbehandlung: Methode einbauen, um Fehler von Workern zum Consumer zu propagieren, z. B. separater Fehlerkanal oder Result-Struktur.
• Gepufferte Kanäle: Können Performance steigern, indem Blockaden reduziert werden – aber sorgfältig testen.

Das Fan-Out/Fan-In-Muster ist ein unverzichtbares Werkzeug im Arsenal jedes Go-Entwicklers für effiziente, skalierbare Datenverarbeitungs-Pipelines – besonders wenn Flexibilität bei Datenflüssen zwischen Stufen gefragt ist.

— Editorial Team