Concurrency Patterns
A few well-known shapes cover most concurrent Go code: a worker pool that processes a queue with a fixed number of goroutines, fan-out/fan-in to spread work and gather the results, a semaphore to cap how many things happen at once, and pipelines of stages. This lesson builds each one from goroutines, channels and sync, with attention to the details that keep them correct: who closes which channel, and how every goroutine finishes.
Worker pool
Starting one goroutine per item is fine for a few hundred items. For a million, or when each job hits a database or API, use a fixed number of workers that take jobs from a shared channel:
package main import ( "fmt" "slices" "strings" "sync" ) type Result struct { Word string Vowels int } func countVowels(w string) int { return strings.Count(w, "a") + strings.Count(w, "e") + strings.Count(w, "i") + strings.Count(w, "o") + strings.Count(w, "u") } func main() { words := []string{"goroutine", "channel", "select", "mutex", "context", "pipeline", "worker"} jobs := make(chan string) results := make(chan Result) const workers = 3 var wg sync.WaitGroup for range workers { wg.Go(func() { for w := range jobs { // each worker takes jobs until the channel closes results <- Result{w, countVowels(w)} } }) } go func() { // feed the jobs, then close so the workers' loops end for _, w := range words { jobs <- w } close(jobs) }() go func() { // close results once every worker has finished wg.Wait() close(results) }() var all []Result for r := range results { all = append(all, r) } slices.SortFunc(all, func(a, b Result) int { return strings.Compare(a.Word, b.Word) }) for _, r := range all { fmt.Printf("%-10s %d\n", r.Word, r.Vowels) } }
channel 2 context 2 goroutine 5 mutex 2 pipeline 4 select 2 worker 2
Three rules make this work. The producer closes jobs when there's no more work. A separate goroutine waits for the workers and then closes results — no worker can close it, because the others might still be sending. And main ranges over results until it's closed. Results arrive in any order, so they're sorted at the end.
Limiting concurrency with a semaphore
Sometimes you're happy to start a goroutine per item but must cap how many run at the same time — no more than 5 downloads at once, say. A buffered channel makes a simple semaphore: each goroutine puts a token in before working and takes it out afterwards, so at most cap can be inside:
package main import ( "fmt" "sync" "sync/atomic" "time" ) func main() { sem := make(chan struct{}, 3) // at most 3 at a time var running, peak atomic.Int32 var wg sync.WaitGroup for range 10 { wg.Go(func() { sem <- struct{}{} // acquire (blocks while 3 are running) defer func() { <-sem }() // release now := running.Add(1) for { p := peak.Load() if now <= p || peak.CompareAndSwap(p, now) { break } } time.Sleep(20 * time.Millisecond) // the "work" running.Add(-1) }) } wg.Wait() fmt.Println("peak concurrency:", peak.Load()) }
peak concurrency: 3
Fan-out, fan-in
Fan-out starts several goroutines reading from one channel; fan-in merges several channels into one. A generic merge is a handy building block:
package main import ( "fmt" "slices" "sync" ) func merge[T any](inputs ...<-chan T) <-chan T { out := make(chan T) var wg sync.WaitGroup for _, in := range inputs { wg.Go(func() { for v := range in { out <- v } }) } go func() { wg.Wait() close(out) }() return out } func squares(nums ...int) <-chan int { out := make(chan int) go func() { defer close(out) for _, n := range nums { out <- n * n } }() return out } func main() { merged := merge(squares(1, 2, 3), squares(4, 5), squares(6)) var got []int for v := range merged { got = append(got, v) } slices.Sort(got) fmt.Println(got) }
[1 4 9 16 25 36]
Collecting errors
When running several tasks where any one failing means the whole operation has failed, you want to know which tasks failed once everything is done. Here each task's error goes into its own slot, and errors.Join reports them all:
package main import ( "errors" "fmt" "sync" ) func check(host string) error { if host == "db.internal" { return fmt.Errorf("%s: connection refused", host) } return nil } func checkAll(hosts []string) error { errs := make([]error, len(hosts)) var wg sync.WaitGroup for i, h := range hosts { wg.Go(func() { errs[i] = check(h) }) } wg.Wait() return errors.Join(errs...) // nil if every slot is nil } func main() { fmt.Println(checkAll([]string{"web.internal", "cache.internal"})) fmt.Println(checkAll([]string{"web.internal", "db.internal"})) }
<nil> db.internal: connection refused
The golang.org/x/sync/errgroup module packages this pattern, adding a concurrency limit and cancelling the remaining tasks via a context as soon as one fails. It's widely used; the next lesson covers the context part.
Pipelines
A pipeline chains stages, each a goroutine connected to the next by a channel (lesson 52 built a small one). Stages run concurrently, so while one item is being written out, the next is already being processed. Combine it with a worker pool when one stage is much slower than the others.
Checklist for concurrent code
- Every goroutine you start has a clear way to finish.
- Exactly one owner closes each channel — the sender, after all sends are done.
- Shared data is either owned by one goroutine or protected by a mutex or atomic.
- Results that need an order are sorted or written to indexed slots.
- Tests run with
-race.
Try it yourself
Use a pool of 4 workers to compute the sum of digits of the numbers 1 to 1000 and add the results together. Change the number of workers — the answer mustn't change.
13501
context.