TopolearnReal concurrent programs are built from a few well-known patterns that also guarantee goroutines exit.
A pipeline is a series of stages connected by channels. Each stage is a function that receives values from an input channel, processes them, and sends to an output channel it owns and closes:
func square(ctx context.Context, in <-chan int) <-chan intTwo rules keep pipelines correct:
- The stage that creates a channel closes it, with
defer close(out)in its goroutine, so downstreamrangeloops end. - Every send also selects on `ctx.Done()`. If the consumer quits early, upstream goroutines must be able to give up instead of blocking forever on a send — otherwise they leak.
Fan-out means starting several goroutines reading from the same input channel to parallelise a slow stage; Go distributes values between them. Fan-in (merge) combines several channels into one: start a goroutine per input that forwards values, and a final goroutine that waits on a WaitGroup and closes the output.
For the very common "run N tasks concurrently, stop on first error" case, use golang.org/x/sync/errgroup:
g, ctx := errgroup.WithContext(ctx);g.Go(func() error {...})for each task.- The first error cancels
ctx, so sibling tasks can stop early;g.Wait()returns that error. g.SetLimit(n)bounds concurrency.
A semaphore — a buffered channel of capacity n where you send to acquire and receive to release — is the simplest way to cap parallelism by hand.
Ordering is lost once you fan out. If results must stay in input order, carry an index with each item and reassemble, or write results into a preallocated slice at that index.
Example
package main
import (
"context"
"fmt"
)
// stage applies f to each value from in; it owns and closes out.
func stage(ctx context.Context, in <-chan int, f func(int) int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
select {
case out <- f(n):
case <-ctx.Done():
return // consumer gave up: exit instead of leaking
}
}
}()
return out
}
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel() // unblocks every stage if we stop early
src := make(chan int, 4)
for i := 1; i <= 4; i++ {
src <- i
}
close(src)
squares := stage(ctx, src, func(n int) int { return n * n })
for v := range stage(ctx, squares, func(n int) int { return n + 1 }) {
fmt.Print(v, " ") // 2 5 10 17
}
}Remember
- Each stage owns and closes its output channel
- Every blocking send also selects on ctx.Done() to prevent leaks
- errgroup covers run-N-tasks-and-stop-on-first-error
Watch out for
- A consumer that stops reading early without cancelling, leaving upstream goroutines blocked forever.