Concurrency Patterns

Go made concurrency approachable — go f() and you're off. It also made it easy to leak goroutines, deadlock, and race. The difference between the two outcomes is knowing a handful of patterns and the rules that keep them safe.

This post covers the three workhorses — pipelines, fan-out/fan-in, and worker pools — and the discipline that makes all of them leak-free. It leans on context for cancellation, so skim that if ctx is unfamiliar.

The two rules that prevent most bugs

Before any pattern, internalize these:

1. Whoever creates a goroutine is responsible for ending it. A goroutine with no exit path is a leak. Every go needs an answer to "how does this stop?"
2. The sender closes the channel, never the receiver. Closing a channel says "no more values are coming," which only the sender can know. Closing from the receive side, or closing twice, panics.

Most concurrency bugs in Go are a violation of one of these. The patterns below are really just disciplined ways to honor them.

Pattern 1: Pipelines

A pipeline is a series of stages connected by channels, each stage a goroutine that receives from an inbound channel, does work, and sends to an outbound one.

// Stage 1: produce numbers.
func gen(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out) // sender closes
        for _, n := range nums {
            select {
            case out <- n:
            case <-ctx.Done(): // honor cancellation
                return
            }
        }
    }()
    return out
}

// Stage 2: square each number.
func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case out <- n * n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

Compose them by passing one stage's output into the next:

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

for n := range square(ctx, gen(ctx, 1, 2, 3, 4)) {
    fmt.Println(n) // 1 4 9 16
}

Note the directional channel types: <-chan int (receive-only) in returns, chan<- where you only send. The compiler enforces the data-flow direction for you — use them; they're free documentation and a real safety net.

The select with ctx.Done() in every send is what makes the pipeline cancellable. Without it, if the consumer stops reading early, the producer blocks forever on out <- and leaks. With it, cancelling the context unwinds every stage cleanly.

Pattern 2: Fan-out / fan-in

When one stage is the bottleneck, fan out: run several copies of it reading from the same input channel. Then fan in: merge their outputs back into one.

// Fan-in: merge multiple channels into one.
func merge[T any](ctx context.Context, cs ...<-chan T) <-chan T {
    out := make(chan T)
    var wg sync.WaitGroup
    wg.Add(len(cs))
    for _, c := range cs {
        go func(c <-chan T) {
            defer wg.Done()
            for v := range c {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }(c)
    }
    // Close out once all input channels are drained.
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

in := gen(ctx, 1, 2, 3, 4, 5, 6, 7, 8)

// Fan out to 3 workers, all reading the same channel.
w1 := square(ctx, in)
w2 := square(ctx, in)
w3 := square(ctx, in)

// Fan in their results.
for n := range merge(ctx, w1, w2, w3) {
    fmt.Println(n)
}

Multiple goroutines reading from the same channel is safe and is exactly how you distribute work — the runtime hands each value to whichever worker is ready. The sync.WaitGroup here solves the "who closes the merged channel?" problem: close it once, after all senders finish.

Pattern 3: Worker pools

Fan-out spins up a goroutine per stage instance. A worker pool is the bounded version: a fixed number of long-lived workers pulling jobs off a shared queue. Use it when you want to cap concurrency — e.g. don't open 10,000 simultaneous connections to a database.

func process(ctx context.Context, jobs <-chan Job, numWorkers int) <-chan Result {
    results := make(chan Result)
    var wg sync.WaitGroup
    wg.Add(numWorkers)

    for i := 0; i < numWorkers; i++ {
        go func() {
            defer wg.Done()
            for job := range jobs { // each worker pulls until jobs is closed
                select {
                case results <- doWork(job):
                case <-ctx.Done():
                    return
                }
            }
        }()
    }

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

The pool size is your concurrency limit. numWorkers tuned to CPU count suits CPU-bound work; for I/O-bound work you can go much higher. (For the common case of "run N tasks with a concurrency cap and collect errors," reach for errgroup with SetLimit from post 6 — it's less boilerplate than a hand-rolled pool.)

The leak you will write at least once

Here's the classic. A function launches a goroutine to do work, the caller times out and moves on, and the goroutine blocks forever trying to send to a channel nobody is reading:

// LEAKY: if the caller stops reading, this goroutine blocks on `ch <-` forever.
func leaky() <-chan int {
    ch := make(chan int)
    go func() {
        ch <- expensiveComputation() // blocks forever if no one receives
    }()
    return ch
}

Two standard fixes:

// Fix A: a buffered channel of size 1 — the goroutine can always send and exit.
func fixedBuffer() <-chan int {
    ch := make(chan int, 1) // buffer absorbs the send
    go func() {
        ch <- expensiveComputation()
    }()
    return ch
}

// Fix B: select on context, so the goroutine exits when the caller gives up.
func fixedContext(ctx context.Context) <-chan int {
    ch := make(chan int)
    go func() {
        select {
        case ch <- expensiveComputation():
        case <-ctx.Done():
        }
    }()
    return ch
}

A buffered channel sized to the number of sends is the simplest cure for fire-and-forget results; context cancellation is the general one.

Detecting problems

• Always run tests with -race. The race detector catches concurrent unsynchronized access. It's the single best tool you have; wire it into CI.
• Watch goroutine counts. A steadily climbing runtime.NumGoroutine() (or the /debug/pprof/goroutine profile) is the signature of a leak.
• Prefer the standard tools to clever channel gymnastics. errgroup, sync.WaitGroup, and a bounded worker pool cover the vast majority of real needs. Elaborate channel choreography is usually a sign you want one of those instead.

The takeaway