Worker Pool in Go: Goroutine-Limits, Ergebnisse und sichere Stornierung

1. Übersicht

Goroutines in Go sind sehr billig, daher ist es üblich, einfach go für jede Aufgabe aufzurufen. Wenn die Zahl der Aufgaben in die Zehntausende geht und jede Aufgabe eine Datenbank oder HTTP aufruft, erschöpft dieses Vorgehen leicht den Connection Pool oder überschreitet das Rate Limit des externen Dienstes.

Dieser Artikel zeigt dir, wie man einen Worker Pool baut – eine feste Anzahl von Goroutines, die Jobs von einem Channel erhalten. Nach dem Lesen wirst du wissen, wie man einen einfachen Pool schreibt, Ergebnisse und Fehler sammelt, ihn mit context abbricht, mit errgroup.SetLimit vereinfacht und testet, dass der Pool tatsächlich die Parallelität begrenzt.

2. Das Problem: unbegrenzte Goroutines

Betrachte diesen Code zur Verarbeitung einer Bestellliste:

for _, o := range orders {
	go chargeOrder(o) // 50.000 Bestellungen -> 50.000 Goroutines rufen Payment API auf
}

Jede Goroutine verbraucht nur wenige KB Speicher, also liegt das Problem normalerweise nicht beim RAM. Das Problem liegt bei den dahinter liegenden Ressourcen:

• Der Connection Pool der Datenbank hat nur etwa 20–50 Verbindungen.
• Das Payment Gateway begrenzt die Zahl der Anfragen pro Sekunde.
• Die Zahl der Datei-Deskriptoren eines Prozesses ist begrenzt.

Der Worker Pool löst das, indem er zwei Konzepte trennt. Die Zahl der Aufgaben kann sehr groß sein, aber die Zahl der gleichzeitig laufenden Aufgaben wird durch die Zahl der Worker begrenzt.

3. Basis-Worker-Pool

Das Modell besteht aus drei Teilen:

1. Ein Channel jobs mit zu erledigenden Aufgaben.
2. N Goroutines als Worker, jeder looped über den Channel.
3. Eine sync.WaitGroup zum Warten auf das Ende aller Worker.

package main

import (
	"fmt"
	"sync"
	"time"
)

type Order struct {
	ID     int
	Amount float64
}

func worker(id int, orders <-chan Order, wg *sync.WaitGroup) {
	defer wg.Done()
	for o := range orders {
		time.Sleep(100 * time.Millisecond) // simuliert Payment API Aufruf
		fmt.Printf("worker %d verarbeitet Bestellung %d (%.0f)\n", id, o.ID, o.Amount)
	}
}

func main() {
	orders := make(chan Order)
	var wg sync.WaitGroup

	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go worker(i, orders, &wg)
	}

	start := time.Now()
	for i := 1; i <= 9; i++ {
		orders <- Order{ID: i, Amount: float64(i) * 10}
	}
	close(orders)
	wg.Wait()
	fmt.Println("fertig nach", time.Since(start).Round(100*time.Millisecond))
}

Der Code funktioniert folgendermaßen:

• orders <-chan Order ist ein Nur-Empfangs-Channel vom Standpunkt des Worker. Der Compiler blockiert den Worker, wenn er versucht, auf den Channel zu schreiben oder ihn zu schließen.
• for o := range orders beendet sich automatisch, wenn der Channel geschlossen ist und keine Elemente mehr vorhanden sind.
• close(orders) ist das Signal “Arbeit erledigt”. Wenn du diese Zeile vergisst, wartet der Worker ewig und wg.Wait() erzeugt einen Deadlock.
• wg.Add(1) muss vor go worker(...) aufgerufen werden, nicht innerhalb der Goroutine.

Beispielausgabe (die Reihenfolge der Zeilen kann unterschiedlich sein):

worker 3 verarbeitet Bestellung 1 (10)
worker 1 verarbeitet Bestellung 2 (20)
worker 2 verarbeitet Bestellung 3 (30)
worker 2 verarbeitet Bestellung 4 (40)
...
fertig nach 300ms

9 Bestellungen geteilt durch 3 Worker, jede Bestellung dauert 100ms, also dauert die Gesamtzeit etwa 300ms statt 900ms mit sequentieller Verarbeitung.

Hinweis: Ab Go 1.25 kannst du wg.Go(func() { worker(i, orders) }) schreiben. Diese Methode ruft automatisch Add und Done auf und vermeidet damit ein Vergessen.

4. Ergebnisse und Fehler sammeln

In der Praxis musst du normalerweise wissen, welche Bestellung erfolgreich war und welche einen Fehler hatte. Dazu fügen wir einen Channel results hinzu:

package main

import (
	"errors"
	"fmt"
	"sync"
)

type Order struct {
	ID     int
	Amount float64
}

type Result struct {
	OrderID int
	Fee     float64
	Err     error
}

func chargeOrder(o Order) (float64, error) {
	if o.Amount <= 0 {
		return 0, errors.New("ungültiger Betrag")
	}
	return o.Amount * 0.02, nil // 2% Gebühr
}

func main() {
	orders := []Order{{1, 100}, {2, 0}, {3, 250}, {4, 80}}

	jobs := make(chan Order)
	results := make(chan Result)
	var wg sync.WaitGroup

	for range 3 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for o := range jobs {
				fee, err := chargeOrder(o)
				results <- Result{OrderID: o.ID, Fee: fee, Err: err}
			}
		}()
	}

	go func() {
		for _, o := range orders {
			jobs <- o
		}
		close(jobs)
	}()

	go func() {
		wg.Wait()
		close(results)
	}()

	for r := range results {
		if r.Err != nil {
			fmt.Printf("Bestellung %d Fehler: %v\n", r.OrderID, r.Err)
			continue
		}
		fmt.Printf("Bestellung %d Gebühr %.2f\n", r.OrderID, r.Fee)
	}
}

Die Ausgabe sieht etwa so aus:

Bestellung 1 Gebühr 2.00
Bestellung 3 Gebühr 5.00
Bestellung 2 Fehler: ungültiger Betrag
Bestellung 4 Gebühr 1.60

Es gibt drei wichtige Punkte zu beachten:

• Das Versenden von Jobs läuft in einer eigenen Goroutine. Wenn main gleichzeitig Jobs versendet und Ergebnisse liest, wird der Worker bei results <- blockiert, während main bei jobs <- blockiert wird. Das Ergebnis ist ein Deadlock.
• Die dritte Goroutine wartet auf wg.Wait() und schließt dann results. Erst wenn alle Worker fertig sind, ist sicher, dass niemand mehr in results schreibt. Die allgemeine Regel: nur die Seite, die sendet, schließt den Channel.
• Fehler werden in Result verpackt statt den ganzen Pool zu stoppen. Der Fehler einer Bestellung sollte andere Bestellungen nicht blockieren.

Die Reihenfolge der Ergebnisse entspricht nicht der Eingabereihenfolge. Wenn du die Reihenfolge erhalten musst, nimm den Index mit im Job oder verwende die Methode aus Abschnitt 6.

5. Pool mit Context stornieren

Wenn eine HTTP-Anfrage storniert wird oder der Timeout überschritten wird, muss der Pool anhalten statt weiterhin die Payment API aufzurufen. Das machst du, indem der Worker auf ctx.Done() hört:

func worker(ctx context.Context, jobs <-chan Order, results chan<- Result) {
	for {
		select {
		case <-ctx.Done():
			return
		case o, ok := <-jobs:
			if !ok {
				return
			}
			fee, err := chargeOrder(o)
			select {
			case results <- Result{OrderID: o.ID, Fee: fee, Err: err}:
			case <-ctx.Done():
				return
			}
		}
	}
}

Die Producer-Seite braucht auch ein ähnliches select:

go func() {
	defer close(jobs)
	for _, o := range orders {
		select {
		case jobs <- o:
		case <-ctx.Done():
			return
		}
	}
}()

Und erstelle ein Context mit Timeout an der Aufrufstelle:

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()

Der Grund für select in beiden Senderichtungen: Wenn der Worker ctx nur beim Empfangen von Jobs prüft, kann er trotzdem bei results <- steckenbleiben, während die lesende Seite bereits weg ist. Das ist eine häufige Quelle für Goroutine-Lecks.

Hinweis: select hilft dem Worker nur, zwischen Jobs auszusteigen. Wenn chargeOrder einen HTTP-Aufruf macht, musst du ctx dort übergeben (http.NewRequestWithContext), um die laufende Anfrage abbrechen zu können.

6. Mit errgroup.SetLimit vereinfachen

Channel, WaitGroup und context von Hand zu schreiben ist fehleranfällig. Das Paket errgroup hat SetLimit, mit dem du einen Pool mit Limit in wenigen Zeilen erstellen kannst. Zuerst die Abhängigkeit hinzufügen:

go get golang.org/x/sync/errgroup

Diese generische Funktion verarbeitet ein Slice mit maximal limit Goroutines, erhält die Ergebnisreihenfolge und stoppt beim ersten Fehler:

package pool

import (
	"context"

	"golang.org/x/sync/errgroup"
)

func Process[T, R any](
	ctx context.Context,
	limit int,
	items []T,
	fn func(context.Context, T) (R, error),
) ([]R, error) {
	results := make([]R, len(items))
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(limit)

	for i, item := range items {
		g.Go(func() error {
			r, err := fn(ctx, item)
			if err != nil {
				return err
			}
			results[i] = r
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return results, nil
}

Der Code funktioniert so:

• g.SetLimit(limit) lässt g.Go blockieren, wenn bereits limit Goroutines laufen. Vom Durchsatz her ist das gleichwertig mit einem Worker Pool.
• errgroup.WithContext gibt ein ctx zurück, das sofort abgebrochen wird, wenn eine Funktion einen Fehler zurückgibt. Andere Aufgaben, die ctx verwenden, bekommen das Stopsignal.
• results[i] = r ist sicher bei gleichzeitiger Ausführung, da jede Goroutine in einen eigenen Index schreibt. Kein Mutex nötig in diesem Fall.
• Ab Go 1.22 werden die Schleifenvariablen i und item bei jeder Iteration neu erstellt, daher hat die Closure nicht mehr das Problem mit gemeinsamen Variablen wie in älteren Versionen.

Der Aufruf sieht so aus:

fees, err := pool.Process(ctx, 5, orders, func(ctx context.Context, o Order) (float64, error) {
	return paymentService.Charge(ctx, o)
})

Der Unterschied zum selbstgeschriebenen Pool: errgroup erstellt eine neue Goroutine für jeden Item statt N Worker wiederverwendet zu werden. Bei den meisten Anwendungen ist dieser Unterschied unbedeutend. Wenn du brauchst, dass Worker einen eigenen Zustand behalten, z. B. jeder Worker hat eine Verbindung, dann schreib den Pool selbst wie in Abschnitt 3–5.

7. Worker Pool testen

Du solltest zwei Hauptverhalten testen. Erstens: Der Pool überschreitet limit niemals. Zweitens: Fehler werden korrekt zurückgegeben. Um die Parallelität zu messen, verwenden wir atomic.Int32:

package pool

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)

func TestProcess_LimitsConcurrencyAndKeepsOrder(t *testing.T) {
	var running, maxSeen atomic.Int32
	items := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

	got, err := Process(context.Background(), 3, items, func(_ context.Context, n int) (int, error) {
		cur := running.Add(1)
		defer running.Add(-1)
		for {
			m := maxSeen.Load()
			if cur <= m || maxSeen.CompareAndSwap(m, cur) {
				break
			}
		}
		time.Sleep(10 * time.Millisecond)
		return n * 2, nil
	})

	if err != nil {
		t.Fatalf("Fehler nicht erwartet: %v", err)
	}
	if maxSeen.Load() > 3 {
		t.Errorf("Gleichzeitig laufend: %d, Limit ist 3", maxSeen.Load())
	}
	want := []int{2, 4, 6, 8, 10, 12, 14, 16, 18, 20}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("got[%d] = %d, want %d", i, got[i], want[i])
		}
	}
}

func TestProcess_ReturnsFirstError(t *testing.T) {
	errDeclined := errors.New("Karte abgelehnt")

	_, err := Process(context.Background(), 2, []int{1, 2, 3}, func(_ context.Context, n int) (int, error) {
		if n == 2 {
			return 0, errDeclined
		}
		return n, nil
	})

	if !errors.Is(err, errDeclined) {
		t.Fatalf("errDeclined erwartet, got %v", err)
	}
}

Teste mit Race Detector:

go test -race ./...

ok  	example.com/shop/pool	0.061s

Die CompareAndSwap-Schleife aktualisiert maxSeen ohne Sperre. Das Flag -race ist bei gleichzeitigem Code Pflicht, da es Zugriffe auf gemeinsame Daten findet, die leicht übersehen werden.

8. Häufige Fehler und wann man keinen Pool braucht

Häufige Fehler:

• close(jobs) vergessen: Worker beendet sich nie, Programm zeigt fatal error: all goroutines are asleep - deadlock!.
• Channel auf der Empfängerseite schließen, oder zweimal schließen: panic: close of closed channel.
• wg.Add innerhalb der Goroutine aufrufen: Wait könnte zurückkehren, bevor die Goroutine Add aufruft.
• Job senden und Ergebnis lesen auf der gleichen Goroutine mit unbufferederm Channel: verursacht Deadlock wie in Abschnitt 4 gezeigt.
• Nicht auf ctx.Done() beim Ergebnis senden achten: führt zu Goroutine-Lecks, wenn die Leseseite schon weg ist.
• Zahl der Worker willkürlich wählen: Bei CPU-gebundenen Aufgaben nutze runtime.NumCPU(). Bei I/O-gebundenen Aufgaben, orientiere dich an den Grenzen der Ressource hinten, z. B. Connection-Pool-Größe.

Wann man keinen Worker Pool braucht:

• Die Zahl der Aufgaben ist klein und bekannt, z. B. 3 Services parallel aufrufen. Hier reicht errgroup ohne SetLimit.
• Aufgaben sind extrem schnell, nur wenige Mikrosekunden. Der Overhead durch den Channel könnte größer sein als die Arbeit selbst; sequentiell oder Batch-Verarbeitung ist effizienter.
• Du musst nach Rate begrenzen (Anfragen/Sekunde), nicht nach Zahl gleichzeitig. Dafür brauchst du einen Rate Limiter, z. B. golang.org/x/time/rate. Der Worker Pool ersetzt ihn nicht.

9. Fazit

Der Worker Pool hilft dir, die Parallelität in Go unter Kontrolle zu halten: Die Zahl der Aufgaben kann riesig sein, aber die Ressourcen dahinter bleiben geschützt. Der Artikel hat diese Schritte durchgegangen:

• Einfacher Pool mit Channel und sync.WaitGroup.
• Ergebnisse und Fehler über einen results-Channel sammeln, Channel an der richtigen Stelle schließen.
• Sicher mit context in beiden Senderichtungen abbrechen.
• Mit errgroup.SetLimit und einer generischen Process-Funktion vereinfachen.
• Parallelitätslimit und Fehlerbehandlung mit go test -race testen.

Für die meisten Fälle ist errgroup.SetLimit die einfachste und fehlerunanfälligste Wahl. Schreib den Pool selbst nur, wenn Worker einen eigenen Zustand brauchen oder wenn du feingranulare Kontrolle über Job- und Ergebnisfluss brauchst.