Worker pools en pipelines in Go: schaalbare concurrency

Als je applicatie duizenden taken moet verwerken zonder je server plat te leggen, zijn worker pools en pipelines in Go je beste vrienden. Beide patronen bouwen voort op goroutines en channels, maar lossen verschillende problemen op: pools beperken parallelisme, pipelines stromen data door opeenvolgende stappen.

In dit artikel leer je hoe je beide patronen praktisch toepast, waar de valkuilen zitten en hoe je ze combineert tot robuuste, schaalbare systemen.

Waarom worker pools en pipelines?

Go maakt concurrency verleidelijk eenvoudig: zet go voor een functie en je hebt een goroutine. Maar ongebreideld goroutines starten, bijvoorbeeld één per inkomend verzoek of één per bestand, leidt tot uitputting van geheugen, database connections of file descriptors.

Een worker pool lost dit op door het aantal gelijktijdige goroutines te beperken. Een pipeline structureert werk in duidelijke stappen waar elke stap onafhankelijk schaalbaar is. Samen vormen ze de basis voor data-intensieve services, ETL-jobs en batch processing.

Als je de basis van goroutines nog wilt opfrissen, lees dan eerst goroutines en concurrency basics en channels diep uitgelegd.

Het worker pool patroon

Een worker pool bestaat uit drie onderdelen: een input channel met taken, een vast aantal workers die taken oppakken, en optioneel een output channel met resultaten.

Basis implementatie

package main

import (
    "fmt"
    "sync"
)

type Job struct {
    ID   int
    Data string
}

type Result struct {
    JobID  int
    Output string
}

func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        output := fmt.Sprintf("worker %d verwerkte %s", id, job.Data)
        results <- Result{JobID: job.ID, Output: output}
    }
}

func main() {
    jobs := make(chan Job, 100)
    results := make(chan Result, 100)

    var wg sync.WaitGroup
    for w := 1; w <= 5; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    go func() {
        for i := 1; i <= 20; i++ {
            jobs <- Job{ID: i, Data: fmt.Sprintf("taak-%d", i)}
        }
        close(jobs)
    }()

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

    for r := range results {
        fmt.Println(r.Output)
    }
}

Merk op hoe close(jobs) de workers signaleert dat er geen werk meer komt. De range in elke worker stopt automatisch zodra het channel leeg en gesloten is.

Het juiste aantal workers kiezen

De vuistregels:

• CPU-gebonden werk: begin met runtime.NumCPU() workers
• I/O-gebonden werk (HTTP calls, database queries): 10–100× het aantal cores
• Gemengd werk: meet met realistische benchmarks

Hardcode nooit een getal zonder te testen. Gebruik performance tuning tools zoals pprof en benchmarks om het optimum te vinden voor jouw workload.

Context en graceful shutdown

Een productie worker pool moet netjes kunnen stoppen. Daarvoor gebruik je context.Context.

func worker(ctx context.Context, id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case <-ctx.Done():
            return
        case job, ok := <-jobs:
            if !ok {
                return
            }
            select {
            case results <- process(job):
            case <-ctx.Done():
                return
            }
        }
    }
}

Met ctx.Done() breekt de worker direct af zodra cancellation wordt gesignaleerd, bijvoorbeeld bij een SIGTERM. Dit patroon zie je ook terug in HTTP servers bouwen in Go, waar graceful shutdown essentieel is.

Pipelines in Go

Een pipeline is een reeks stappen die via channels met elkaar verbonden zijn. Elke stap is een goroutine (of groep goroutines) die input uit één channel leest en output naar een volgend channel schrijft.

Het klassieke Go Blog artikel over pipelines van het Go team beschrijft dit patroon uitgebreid.

Een drie-stap pipeline

Stel, je leest URL's in, downloadt de content en slaat de grootte op:

func generate(ctx context.Context, urls []string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for _, u := range urls {
            select {
            case out <- u:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func fetch(ctx context.Context, in <-chan string) <-chan []byte {
    out := make(chan []byte)
    go func() {
        defer close(out)
        for url := range in {
            resp, err := http.Get(url)
            if err != nil {
                continue
            }
            body, _ := io.ReadAll(resp.Body)
            resp.Body.Close()
            select {
            case out <- body:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func measure(ctx context.Context, in <-chan []byte) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for body := range in {
            select {
            case out <- len(body):
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

Je zet de pipeline op met: sizes := measure(ctx, fetch(ctx, generate(ctx, urls))). Elke stap sluit zijn eigen output channel zodra input op is, downstream stops cascaderen automatisch.

Fan-out en fan-in

Een enkele stap kan een bottleneck vormen. Met fan-out splits je het werk over meerdere goroutines; met fan-in combineer je hun resultaten.

func fanOut(ctx context.Context, in <-chan string, n int) []<-chan []byte {
    outs := make([]<-chan []byte, n)
    for i := 0; i < n; i++ {
        outs[i] = fetch(ctx, in)
    }
    return outs
}

func fanIn(ctx context.Context, chans ...<-chan []byte) <-chan []byte {
    out := make(chan []byte)
    var wg sync.WaitGroup
    wg.Add(len(chans))
    for _, c := range chans {
        go func(c <-chan []byte) {
            defer wg.Done()
            for v := range c {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }(c)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

Nu verwerken meerdere fetch goroutines parallel URL's uit hetzelfde input channel, terwijl de output als één stroom doorgaat naar de volgende stap.

Backpressure en buffered channels

Zonder backpressure produceert een snelle stap meer data dan een trage volgende stap aankan, geheugen loopt op, latency stijgt. Er zijn twee oplossingen:

Unbuffered channels dwingen synchronisatie af: de producer blokkeert tot de consumer klaar is om te ontvangen. Dit is de eenvoudigste vorm van backpressure.

Buffered channels bieden een kleine buffer die korte pieken opvangt: make(chan Job, 10). Kies de buffergrootte bewust, te groot verbergt problemen, te klein veroorzaakt onnodige blocking.

In een worker pool geeft het input channel automatisch backpressure: als alle workers bezig zijn en de buffer vol is, blokkeert de producer tot er ruimte vrijkomt.

Errors afhandelen in pipelines

Errors in een pipeline zijn lastig omdat je geen simpele return err kunt doen: de goroutines draaien onafhankelijk. Drie opties:

1. Error channel: elke stap schrijft errors naar een apart channel dat de main goroutine leest
2. Result wrapper: combineer value en error in één struct die door het channel gaat
3. errgroup: gebruik golang.org/x/sync/errgroup voor automatische cancellation bij de eerste fout

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

g, ctx := errgroup.WithContext(ctx)
for i := 0; i < 5; i++ {
    g.Go(func() error {
        return worker(ctx, jobs, results)
    })
}
if err := g.Wait(); err != nil {
    log.Printf("pipeline error: %v", err)
}

errgroup cancelt de context zodra één goroutine een error retourneert, alle andere workers stoppen dan automatisch. Voor meer over idiomatische foutafhandeling zie functions en error handling.

Praktijkvoorbeeld: batch database import

Stel, je importeert 100.000 records uit een CSV in een database. Zonder pool open je potentieel 100.000 connections, je database valt om.

func importCSV(ctx context.Context, db *sql.DB, file string) error {
    records := make(chan []string, 100)
    g, ctx := errgroup.WithContext(ctx)

    g.Go(func() error {
        defer close(records)
        return readCSV(ctx, file, records)
    })

    for i := 0; i < 10; i++ {
        g.Go(func() error {
            for rec := range records {
                if err := insertRecord(ctx, db, rec); err != nil {
                    return err
                }
            }
            return nil
        })
    }

    return g.Wait()
}

Tien workers tegelijk, buffer van 100 records, automatische cancellation bij fouten. Dit patroon combineert soepel met database toegang in Go en goede logging en observability om de voortgang te monitoren.

Valkuilen om te vermijden

• Vergeten channels te sluiten: downstream range blijft dan voor altijd wachten
• Dubbele close: close() op een al gesloten channel paniekt, sluit alleen aan de producer-kant
• Goroutine leaks: als een worker blokkeert op een channel zonder context-check, stopt hij nooit
• Te kleine buffers zonder reden: meet eerst, optimaliseer daarna
• Tests vergeten: pipelines zijn notoir lastig te debuggen zonder goede tests, zie testing in Go voor tabletests en race detection met go test -race

Veelgestelde vragen

Wat is een worker pool in Go?

Een worker pool is een vast aantal goroutines dat taken uit een gedeelde queue verwerkt. Dit beperkt het aantal gelijktijdige bewerkingen en voorkomt dat je systeem overbelast raakt bij grote hoeveelheden werk.

Wanneer gebruik je een pipeline in plaats van een worker pool?

Een pipeline gebruik je wanneer je data in meerdere opeenvolgende stappen verwerkt, zoals lezen, transformeren en opslaan. Een worker pool is ideaal voor onafhankelijke taken die parallel uitgevoerd kunnen worden. In de praktijk combineer je beide: een pipeline waarvan één of meer stappen intern een worker pool hebben.

Hoeveel workers moet een pool hebben?

Voor CPU-intensief werk is runtime.NumCPU() een goed startpunt. Voor I/O-gebonden werk kan het aantal veel hoger liggen, vaak tien tot honderd keer het aantal CPU-cores. Meet altijd met benchmarks voordat je dit vastlegt.

Wat is backpressure en waarom is het belangrijk?

Backpressure zorgt dat een snelle producer niet meer data produceert dan de consumer kan verwerken. Dit voorkom je in Go met buffered channels of een worker pool met een beperkte queue-grootte. Zonder backpressure loopt je geheugengebruik op en wordt je applicatie onvoorspelbaar.

Hoe stop je een worker pool netjes?

Gebruik een context.Context om cancellation te signaleren en close() op het input channel zodra geen werk meer volgt. Een sync.WaitGroup of errgroup.Group wacht tot alle workers hun huidige taak hebben afgerond voordat je main goroutine exit.

Afronding

Worker pools en pipelines zijn twee van de krachtigste patronen in Go's concurrency toolkit. Met beperkte goroutines, buffered channels, context cancellation en errgroup bouw je systemen die duizenden taken per seconde verwerken zonder onvoorspelbaar gedrag.

Begin eenvoudig: eerst een werkende pool, dan meten, dan fan-out toevoegen waar nodig. Over-engineering is makkelijker te maken dan ongedaan, dus laat de profiler je keuzes sturen.