Worker pool — Aufgabenwarteschlange + N Worker

1. Wozu ein worker pool gebraucht wird

Wenn man zum ersten Mal von goroutine erfährt, zieht das Gehirn eine naheliegende Schlussfolgerung: „Prima! Dann eine goroutine pro Aufgabe.“ Das ist ein bisschen wie der Moment, in dem jemand zum ersten Mal von unendlicher Pasta gehoert hat: Man will sofort testen, ob sie wirklich unendlich ist. Das Problem ist, dass die Unendlichkeit normalerweise beim Wort memory endet.

Wenn es nur wenige Aufgaben gibt (zum Beispiel 10–100), funktioniert der Ansatz „eine goroutine pro Aufgabe“ oft tatsächlich und sieht schön aus. Wenn es aber viele Aufgaben sind (Tausende, Hunderttausende), erzeugt man ploetzlich eine riesige Armee von goroutine, die:

• erstens einen Overhead hat (wenn auch kleinen, aber nicht null);
• zweitens schwerer korrekt zu beenden ist (vor allem, wenn es Ergebniskanäle gibt);
• drittens die Kontrollierbarkeit verliert: „Wie viele Aufgaben laufen gerade gleichzeitig?“ wird dann zur Glaubensfrage.

Ein Worker pool löst genau das: Wir legen eine feste Zahl von Workern an (zum Beispiel 4, 8, 16), die Aufgaben aus einer gemeinsamen Warteschlange nehmen und verarbeiten. Das ist ungefähr wie Kassen im Supermarkt: Es kann viele Kundinnen und Kunden geben, aber die Zahl der Kassen ist begrenzt, und die Schlange reguliert die Last.

2. Rollen und Protokoll des worker pool

Bevor man Code schreibt, ist es wichtig, nicht kopfüber in make(chan ...) zu springen, sondern die Rollen zu vereinbaren. Im worker pool ist es meist praktisch, mit mehreren „Figuren“ zu denken: Jemand legt Aufgaben in die Warteschlange, jemand nimmt sie heraus, jemand schließt den Ausgang, jemand sammelt das Ergebnis.

Unten ist ein kleines Schema, um das Bild festzuhalten (wir kommen später darauf zurück, wenn wir alles zu einem Programm verbinden):

flowchart LR
    P[Producer<br/>creates tasks] -->|jobs channel| W1[Worker 1]
    P -->|jobs channel| W2[Worker 2]
    P -->|jobs channel| WN[Worker N]

    W1 -->|results channel| C[Collector<br/>collects results]
    W2 -->|results channel| C
    WN -->|results channel| C

Die wichtigste Regel, die vor der Hälfte aller Paniken und Hänger bewahrt:

Den Kanal schließt derjenige, der in ihn schreibt (der Besitzer des Sendens), und zwar genau einmal.

Wenn es mehrere Schreiber gibt (mehrere Worker schreiben in results), dann übernimmt das Schließen eine separate Instanz (ein Koordinator), die keine Ergebnisse schreibt, sondern nur auf das Ende der Worker wartet und dann den Kanal schließt. Diese Regel hilft dabei, das klassische panic: close of closed channel zu vermeiden.

Zu close ist es nützlich, sich die Grundbedeutung zu merken: Das Schließen eines Kanals bedeutet „Es werden keine Werte mehr gesendet“. Da gibt es keine „Magie des Löschens von Daten“: Bereits gesendete Werte kann man weiterhin auslesen.

3. Wir bereiten ein Beispiel vor: Task, Job und Result

Damit die Beispiele nicht abstrakt als „job #123“ daherkommen, bleiben wir beim Stil einer fiktiven Anwendung „Aufgabenliste“. Stellen wir uns vor, wir haben Task und wollen Statistiken zu den Titeln berechnen: zum Beispiel die Anzahl der Wörter in Title. Die Aufgabe ist absichtlich einfach: Wir brauchen kein Dateisystem, kein HTTP, keine Datenbanken — nur parallele Verarbeitung.

Task-Modell

package main

type Task struct {
	ID    int
	Title string
	Done  bool
}

Die Typen TaskJob und TaskResult

Wir definieren, was „Arbeit“ und was „Ergebnis“ ist. Wichtig: Wir führen separate Typen nicht ein, weil Go das erzwingt, sondern wegen der Lesbarkeit. Ein Kanal chan int ohne Namen ist ein Rätsel. Ein Kanal chan TaskJob ist schon fast Dokumentation.

package main

type TaskJob struct {
	Task Task
}

type TaskResult struct {
	TaskID    int
	WordCount int
	Err       error
}

Achten Sie darauf: TaskResult enthält Err error. Heute bauen wir keine komplexe Fehlerstrategie, aber schon jetzt ist es nützlich, sich daran zu gewöhnen, dass das Ergebnis einer Arbeit „Erfolg oder Fehler“ sein kann, und dass man das bequem in einem Objekt uebergeben kann.

Fehler ErrEmptyTitle

package main

import "errors"

var ErrEmptyTitle = errors.New("empty title")

4. Einfache Versionen ohne worker pool

Jede Form von Parallelitaet wird verständlicher, wenn man zuerst die normale sequenzielle Variante sieht. Und danach, was Kanäle und goroutine daraus machen.

Sequenzielle Variante

package main

import (
	"fmt"
	"strings"
)

func wordCount(title string) int {
	return len(strings.Fields(title))
}

func main() {
	tasks := []Task{
		{ID: 1, Title: "buy milk"},
		{ID: 2, Title: "write go code"},
	}

	for _, t := range tasks {
		fmt.Println(t.ID, wordCount(t.Title)) // 1 2; 2 3 (die Reihenfolge ist stabil)
	}
}

Die sequenzielle Variante liefert stabile Reihenfolge, Einfachheit und minimalen Detailaufwand. Wenn die Verarbeitung jedoch schwer ist (zum Beispiel Parsing, Hashing, Validierung eines großen Imports), kann die Sequenz langsam sein. Und dann entsteht der Wunsch, parallel auszuführen.

Naive Parallelität: eine goroutine pro Aufgabe

Jetzt schreiben wir eine naive Variante: Wir starten fuer jede Task eine goroutine. Das sieht hübsch aus… genau bis zu dem Moment, in dem es „viele“ Aufgaben werden.

package main

import (
	"fmt"
	"strings"
	"sync"
)

func main() {
	tasks := []Task{{ID: 1, Title: "buy milk"}, {ID: 2, Title: "write go code"}}

	var wg sync.WaitGroup
	for _, t := range tasks {
		wg.Add(1)
		go func(task Task) {
			defer wg.Done()
			n := len(strings.Fields(task.Title))
			fmt.Println(task.ID, n)
		}(t)
	}
	wg.Wait()
}

Hier taucht schon eine wichtige Anfängerfalle auf: Wir übergeben t als Argument task vom Typ Task, damit wir nicht in die Falle der Schleifenvariable geraten. Das haben Sie schon frueher gesehen: Selbst in Go 1.22+ gibt es Szenarien, in denen man versehentlich eine Variable zwischen Iterationen teilt, wenn sie nicht dort deklariert wurde, wo sie sein sollte.

Aber das Hauptproblem ist nicht das. Das Problem ist, dass Sie, wenn tasks zum Beispiel 200_000 wird, 200_000 goroutine erzeugen. Sie mögen leichtgewichtig sein, aber nicht leichtgewichtig genug, um sich darüber wie über ein Fest zu freuen. Ein Worker pool macht diese Konstruktion beherrschbar.

5. Implementierung des worker pool: der Code Schritt für Schritt

Jetzt setzen wir die Schlüsselelemente zusammen: Worker, Aufgabenwarteschlange, Koordinator für das Schließen von results, Producer und Collector, und danach main, der alles verbindet.

Worker: liest aus jobs und schreibt in results

Das zentrale Element ist der Worker. Er soll Aufgaben aus jobs nehmen, bis der Kanal geschlossen wird. Das passt perfekt zu for range, weil range über einen Kanal endet, wenn der Kanal geschlossen ist und alle Werte gelesen wurden.

Achten Sie auf die Signatur: jobs <-chan TaskJob (nur lesen), results chan<- TaskResult (nur schreiben). Gerichtete Kanäle sind eine kleine, aber sehr nützliche „Gehirn-Sicherung“: Die Funktion kann nicht versehentlich den fremden Kanal schließen oder dort lesen, wo sie es nicht soll.

package main

import (
	"strings"
	"sync"
)

func worker(jobs <-chan TaskJob, results chan<- TaskResult, wg *sync.WaitGroup) {
	defer wg.Done()

	for j := range jobs {
		title := j.Task.Title
		if title == "" {
			results <- TaskResult{TaskID: j.Task.ID, Err: ErrEmptyTitle}
			continue
		}

		n := len(strings.Fields(title))
		results <- TaskResult{TaskID: j.Task.ID, WordCount: n}
	}
}

Wichtig: Der Worker schließt results nicht. Er ist nicht der „Besitzer des Ergebniskanals“, weil es mehrere Worker gibt. Wenn jeder Worker versucht, results zu schließen, entsteht ein Rennen darum, wer zuerst schließt, und die anderen bekommen eine Panik.

Starten des Pools: N Worker und WaitGroup erzeugen

Ein Worker pool beginnt mit der Warteschlange, aber technisch ist es oft bequemer, zuerst den „Starter“ der Worker zu schreiben.

package main

import "sync"

func startPool(n int, jobs <-chan TaskJob, results chan<- TaskResult) *sync.WaitGroup {
	var wg sync.WaitGroup
	for i := 0; i < n; i++ {
		wg.Add(1)
		go worker(jobs, results, &wg)
	}
	return &wg
}

Beachten Sie die Disziplin der WaitGroup: Add(1) wird vor dem Start der goroutine ausgeführt. Das ist keine Bürokratie — das ist die Methode, um kein seltenes, aber sehr unangenehmes Rennen zu erwischen, bei dem die goroutine schon beendet ist und Done() aufgerufen hat, während der Zähler noch nicht erhöht wurde.

Koordinator: results nach dem Ende der Worker schließen

Collector liest for r := range results { ... }. Damit die Schleife endet, muss der Kanal results geschlossen werden. Aber „wenn es soweit ist“ kann ihn nur derjenige schließen, der weiß, dass alle Worker fertig sind.

Das ist die Rolle des Koordinators: wg.Wait() abwarten und danach close(results) aufrufen.

package main

func closeResultsWhenDone(wg *sync.WaitGroup, results chan<- TaskResult) {
	go func() {
		wg.Wait()
		close(results)
	}()
}

Warum schließen wir results in einer separaten goroutine? Weil man sich sonst mit wg.Wait() direkt in main frühzeitig blockieren kann und beispielsweise nicht mehr rechtzeitig mit dem Lesen der Ergebnisse beginnt. In einfachen Beispielen ist das nicht immer kritisch, aber als Gewohnheit ist es sehr nützlich: „Der Koordinator lebt separat und schließt, wenn die Arbeit erledigt ist.“

Producer: Aufgaben senden und jobs schließen

Der Producer legt Aufgaben in die Warteschlange. Hier gilt das Prinzip des Eigentums: Wenn der Producer in jobs schreibt, dann schließt der Producer auch jobs.

package main

func produceTasks(tasks []Task, jobs chan<- TaskJob) {
	defer close(jobs)

	for _, t := range tasks {
		jobs <- TaskJob{Task: t}
	}
}

defer close(jobs) ist ein guter Trick: Selbst wenn man später einen frühen Ausstieg oder Fehlerbehandlung hinzufügt, ist die Wahrscheinlichkeit deutlich kleiner, den Kanal zu vergessen.

Collector: results bis zum Schließen lesen und aggregieren

Der Collector liest results und macht etwas Sinnvolles damit: ausgeben, in einen Slice speichern, einen Bericht erzeugen.

Machen wir eine einfache Sammlung: Wir geben alles aus und zaehlen, wie viele Fehler es gab.

package main

import "fmt"

func collect(results <-chan TaskResult) int {
	errCount := 0
	for r := range results {
		if r.Err != nil {
			fmt.Println("error:", r.TaskID, r.Err) // error: 3 empty title
			errCount++
			continue
		}
		fmt.Println("ok:", r.TaskID, r.WordCount)
	}
	return errCount
}

Hier ist es wichtig, den folgenden Punkt innerlich zu akzeptieren: Die Reihenfolge der Ergebnisse ist nicht garantiert. Worker Nummer 2 kann Aufgabe Nummer 10 schneller verarbeiten als Worker Nummer 1 Aufgabe Nummer 1. Wenn Sie eine stabile Reihenfolge brauchen, lösen Sie das separat (meist über Sortieren oder indem das Ergebnis nach Index abgelegt wird), aber das ist bereits eine naechste Ebene der Logik.

Alles in main zusammenbauen

main macht fünf Dinge:

• Kanäle erzeugen,
• Worker starten,
• den Koordinator zum Schließen von results starten,
• den Producer starten,
• die Ergebnisse lesen.

package main

import "fmt"

func main() {
	tasks := []Task{
		{ID: 1, Title: "buy milk"},
		{ID: 2, Title: "write go code"},
		{ID: 3, Title: ""},
	}

	jobs := make(chan TaskJob)
	results := make(chan TaskResult)

	wg := startPool(2, jobs, results)
	closeResultsWhenDone(wg, results)

	go produceTasks(tasks, jobs)

	errCount := collect(results)
	fmt.Println("errors:", errCount) // errors: 1
}

Achten Sie auf mehrere Punkte, die zusammen das „Schema ohne Hänger“ ergeben:

• produceTasks wird in einer goroutine gestartet, damit main nicht beim Senden der Aufgaben stehen bleibt und parallel mit dem Sammeln der Ergebnisse beginnen kann.
• jobs wird vom Producer geschlossen, und die Worker beenden sich natürlich: for range jobs endet.
• results wird vom Koordinator geschlossen, der auf alle Worker gewartet hat.
• Der Collector liest results über range, und die Schleife endet automatisch nach dem Schließen des Kanals.

Das ist das „Protokoll“ des worker pool: Niemand wartet auf eine „magische Zahl von Ergebnissen“, niemand schließt Kanäle „nach Augenmaß“, das Ende ergibt sich aus dem Vertrag.

Warum das Schema nicht hängt

Wenn Anfänger zum ersten Mal einen worker pool sehen, denken sie meist: „Hier gibt es doch jede Menge Blockierungen: Kanäle, Warten, Goroutinen… warum hängt das überhaupt nicht?“ Die ruhige Antwort lautet: An jeder blockierenden Stelle gibt es einen garantierten „Entblocker“.

• Der Producer kann an jobs <- ... hängenbleiben, aber die Worker lesen jobs und entlasten die Warteschlange.
• Ein Worker kann an results <- ... hängenbleiben, aber der Collector liest results.
• Der Collector kann „ewig warten“, aber der Koordinator schließt results, wenn die Worker fertig sind.
• Worker können beim Lesen von jobs „ewig warten“, aber der Producer schließt jobs.

Das Ende folgt also nicht einem Timer und auch nicht dem Zufall, sondern logisch aus dem Schließen der Kanäle und der WaitGroup.

6. Typische Fehler bei der Arbeit mit worker pool

Fehler Nr. 1: Man vergisst, jobs zu schließen.
Dann bleiben die Worker fuer immer in for range jobs, weil range über einen Kanal erst nach close(jobs) endet. Und wenn die Worker nicht fertig werden, schließt der Koordinator results nicht, und der Collector wartet ebenfalls unendlich. Das sieht meist so aus, als ob „das Programm nichts tut und nicht endet“, und man beginnt zu vermuten, dass der Computer beleidigt ist.

Fehler Nr. 2: Man schließt results innerhalb des Workers.
Das führt fast garantiert zu panic: close of closed channel, weil es mehrere Worker gibt und jeder sich für den Wichtigsten hält. Wenn das Schließen eines Kanals vom Abschluss mehrerer goroutine abhängt, schließt ihn nicht „irgendein Teilnehmer“, sondern der Koordinator nach wg.Wait().

Fehler Nr. 3: Man macht wg.Add(1) nach go worker(...).
Manchmal „funktioniert“ das, und manchmal bekommt man eine Panik wie sync: negative WaitGroup counter oder seltsame Hänger. Der Grund ist, dass die goroutine eventuell schon Done() aufrufen konnte, während man Add(1) noch nicht ausgeführt hat. Die Disziplin ist einfach: Add immer vor dem Start der goroutine.

Fehler Nr. 4: Der Collector liest nicht per range results, sondern genau N-mal.
Solcher Code ist fragil: Man ändert die Zahl der Aufgaben, fügt einen frühen Ausstieg im Producer hinzu, ergänzt Filterung — und plötzlich liest man entweder Ergebnisse nicht vollständig oder wartet auf zu viele. Das Lesen per range über results knüpft an das reale Beendigungsprotokoll an und nicht an Annahmen.

Fehler Nr. 5: Man erwartet, dass die Ergebnisse in derselben Reihenfolge kommen wie die Aufgaben.
Im worker pool ist die Ausführungsreihenfolge nicht deterministisch. Wenn Sie eine Ordnung brauchen, müssen Sie sie explizit wiederherstellen: Zum Beispiel die Ergebnisse nach TaskID in einem Slice/Map sammeln und sie dann in der gewünschten Reihenfolge ausgeben. Wichtig ist, nicht gegen die Parallelität anzukämpfen, indem man stabile Reihenfolge erwartet, sondern die Reihenfolge als eigene Formatierungsaufgabe zu betrachten.