Go-concurrency zonder leaks: workers, backpressure en eigenaarschap

Een goroutine starten is makkelijk. Bepalen wie erop wacht, wie hem kan stoppen en wat er gebeurt wanneer de consumer vertrekt is het echte werk. Bruikbare Go-concurrency is meestal kleiner dan het eerste ontwerp: een vaste hoeveelheid parallel werk, een zichtbare queue-policy en één eigenaar per channel zijn genoeg voor veel API's, importers en achtergrondtaken.

Bewijs eerst dat concurrency helpt

Ik houd de eerste implementatie synchroon wanneer die de vereiste latency of throughput al haalt. Concurrency voegt scheduling, cancellation, ordering en gedeeltelijke fouten toe. Dat verdient de kosten wanneer onafhankelijk werk op I/O wacht of CPU-werk meerdere cores kan gebruiken, niet alleen omdat goroutines goedkoop zijn.

Voor workers benoem ik de beperkte resource: databaseverbindingen, upstream rate limit, geheugen per job, CPU of file descriptors. Het aantal workers volgt die grens. Vijftig goroutines voor een databasepool van tien verbindingen leveren meestal wachttijd en slechtere tail latency op.

Iedere goroutine heeft een eigenaar en uitgang nodig

Bij iedere go-statement wil ik drie antwoorden: wie wacht erop, welk signaal stopt hem en wat gebeurt er wanneer send of receive niet kan afronden? Een goroutine die eeuwig op een verlaten channel wacht is een leak, ook zonder CPU-gebruik. Hij kan requestdata, sockets, timers en andere objecten vasthouden.

Channel-eigenaarschap is eveneens expliciet. De producer sluit normaal het channel na de laatste send; receivers sluiten het niet om te melden dat zij klaar zijn. Cancellation reist apart via context. Een select op ctx.Done geeft een geblokkeerde send of receive een uitgang.

func produce(ctx context.Context, values []Item) <-chan Item {
    out := make(chan Item)

    go func() {
        defer close(out)
        for _, item := range values {
            select {
            case out <- item:
            case <-ctx.Done():
                return
            }
        }
    }()

    return out
}

Begrens actief werk en wachtend werk afzonderlijk

Een vast aantal workers begrenst actief werk, maar een enorm gebufferd channel verplaatst het probleem naar geheugen. Queuecapaciteit is onderdeel van het servicecontract. Wanneer de buffer vol is, moet de producer blokkeren, weigeren, samenvoegen of duurzaam opslaan. Een onbeperkte geheugenqueue is vooral een uitgestelde storing.

Blokkeren past wanneer de caller veilig kan vertragen, zoals een batchreader voor een parser. Weigeren past bij een live API die 429 of 503 kan teruggeven. Duurzame opslag past bij werk dat een procesrestart moet overleven. Ik maak die keuze zichtbaar in code en metrics.

• Workers begrenzen gelijktijdig resourcegebruik
• De channelbuffer begrenst wachten in geheugen
• De enqueue-deadline begrenst wachttijd van producers
• Geweigerd of opgeslagen werk heeft een expliciet caller-contract nodig
• Queuediepte en leeftijd van de oudste job tonen druk vroeg

Gebruik errgroup wanneer sibling-taken samen slagen of falen

golang.org/x/sync/errgroup is nuttig wanneer meerdere goroutines bij één operatie horen. WithContext annuleert de afgeleide context zodra één functie een fout geeft en Wait retourneert de eerste fout nadat alles is gestopt. SetLimit begrenst actieve goroutines en is voor een eindige lijst vaak eenvoudiger dan een eigen workerpool.

De functies moeten de afgeleide context wel gebruiken. errgroup kan geen databasecall of HTTP-request onderbreken die een andere context kreeg. De caller die de group maakt blijft ook eigenaar van Wait; dat houdt de levenscyclus begrijpelijk.

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(8)

for _, accountID := range accountIDs {
    g.Go(func() error {
        return refreshAccount(ctx, accountID)
    })
}

if err := g.Wait(); err != nil {
    return fmt.Errorf("refresh accounts: %w", err)
}

Een langlevende workerpool heeft een andere levenscyclus

Voor een procesbrede pool start ik workers in één component, sluit ik input pas wanneer geen producer meer kan sturen, annuleer ik de context tijdens shutdown en wacht ik op iedere worker. Jobs krijgen de poolcontext en kunnen daar eventueel een kortere eigen deadline van afleiden.

Shutdownbeleid moet eerlijk zijn. Een service kan geaccepteerde jobs afmaken, direct stoppen en elders retryen of alleen duurzaam opgeslagen werk voltooien. Een willekeurige geheugenqueue gegarandeerd binnen een vaste grace period leegmaken kan niet. Werk dat nooit verloren mag gaan hoort vóór acknowledgement in duurzame opslag.

Test beëindiging en niet alleen resultaten

Een concurrencytest dekt vroeg stoppen van de consumer, een falende worker, geannuleerde parent-context, volle queue en shutdown met actief werk. Begrensde testdeadlines zorgen dat een regressie faalt in plaats van de hele suite te laten hangen. De race detector is waardevol, maar bewijst niet dat goroutines eindigen.

Ik test observeerbare completion in plaats van globale goroutine-aantallen. Runtime en libraries hebben eigen goroutines. Een WaitGroup, errgroup of gesloten done-channel kan aantonen dat de goroutines waarvan de test eigenaar is werkelijk terugkwamen.

Praktische checklist

• Benoem de resource die concurrency moet beschermen
• Geef iedere goroutine een cancel- en join-pad
• Laat producers hun outputchannel sluiten
• Begrens actief en queued werk apart
• Kies bewust tussen blokkeren, weigeren en opslaan
• Test vroeg stoppen, cancellation en shutdown
• Draai de race detector voor gewijzigde concurrente code