Concurrency Patterns

Goroutines and channels are primitives. A handful of patterns built from them cover most real concurrent code. Each one answers the same questions: who closes which channel, and how does everything stop?

Worker pool

A fixed number of goroutines pull jobs from a shared channel. This bounds concurrency — useful when each job hits a rate-limited API or a database with a limited connection pool.

package main

import (
    "fmt"
    "sort"
    "sync"
)

func worker(jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for n := range jobs {
        results <- n * n
    }
}

func main() {
    jobs := make(chan int)
    results := make(chan int)

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

    // Close results once every worker has finished.
    go func() {
        wg.Wait()
        close(results)
    }()

    // Feed jobs, then close jobs so workers' range loops end.
    go func() {
        for i := 1; i <= 5; i++ {
            jobs <- i
        }
        close(jobs)
    }()

    var squares []int
    for sq := range results {
        squares = append(squares, sq)
    }
    sort.Ints(squares)
    fmt.Println(squares)
}

Expected output:

[1 4 9 16 25]

Results arrive in whatever order workers finish, so the program sorts them before printing. The ownership rule: the sender closes. The feeder closes jobs; a separate goroutine closes results after wg.Wait(), because no single worker knows when it's the last one.

Pipeline

A pipeline is a chain of stages, each a goroutine that reads from one channel and writes to the next. Each stage owns and closes its output channel.

package main

import "fmt"

func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

func main() {
    for v := range square(square(generate(1, 2, 3))) {
        fmt.Println(v)
    }
}

Expected output:

1
16
81

With a single goroutine per stage, order is preserved.

Fan-out, fan-in

Fan-out starts several goroutines reading from the same input channel. Fan-in merges their output channels into one:

func merge(chans ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, c := range chans {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(c)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

// Fan out: two square stages share one input. Fan in: merge them.
// in := generate(1, 2, 3, 4)
// for v := range merge(square(in), square(in)) { ... }

Merged output order is not deterministic.

In real code each stage should also take a ctx and select on ctx.Done() when sending, so a consumer that stops early doesn't leave upstream goroutines blocked forever.

errgroup

Hand-rolling a WaitGroup plus error collection plus cancellation gets repetitive. The golang.org/x/sync/errgroup package (maintained by the Go team, but a separate module) packages it up:

go get golang.org/x/sync/errgroup

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

func fetchAll(ctx context.Context, urls []string) error {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(4) // at most 4 in flight

    for _, url := range urls {
        g.Go(func() error {
            req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
            if err != nil {
                return err
            }
            resp, err := http.DefaultClient.Do(req)
            if err != nil {
                return err
            }
            return resp.Body.Close()
        })
    }
    return g.Wait() // first non-nil error; ctx is cancelled when one fails
}

Since Go 1.22 each loop iteration gets its own url variable, so capturing it in the closure is safe.

Common mistakes

• Closing a channel from the receiving side, or from more than one sender — sending on a closed channel panics.
• Forgetting to close a channel that someone ranges over — the loop never ends and the program deadlocks or leaks a goroutine.
• Starting one goroutine per item for an unbounded input (millions of rows). Use a worker pool or SetLimit.
• Reaching for channels when a mutex is simpler. Channels are for passing ownership of data; a mutex is fine for protecting a shared counter or map.

Practice

Using a worker pool of 4 goroutines, compute the total length of these words concurrently and print it: "go", "channels", "select", "mutex", "context".

package main

import (
    "fmt"
    "sync"
)

func main() {
    words := []string{"go", "channels", "select", "mutex", "context"}
    jobs := make(chan string)
    lengths := make(chan int)

    var wg sync.WaitGroup
    for i := 0; i < 4; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for w := range jobs {
                lengths <- len(w)
            }
        }()
    }
    go func() {
        for _, w := range words {
            jobs <- w
        }
        close(jobs)
    }()
    go func() {
        wg.Wait()
        close(lengths)
    }()

    total := 0
    for n := range lengths {
        total += n
    }
    fmt.Println(total)
}

28