Go Concurrency Patterns: A Practical Catalogue
Reviewer’s note: This is a catalogue of the concurrency patterns you actually reach for in Go services, each with a runnable example and a note on when to use it versus something simpler. It is for developers who can already start a goroutine and read a channel, and want a reference they can lift code from. Every example was run on Go 1.24.7. Works with Go 1.21+.
If goroutines or channels are still shaky, the complete Go tutorial covers the language, the goroutines concurrency guide covers the runtime, and Go channels explained covers the one primitive every pattern below is built from. The canonical sources for this material are Rob Pike’s talk Go Concurrency Patterns and the Go blog post Go Concurrency Patterns: Pipelines and cancellation; this article turns those ideas into copy-paste code with production framing.
The two rules every pattern here obeys
Before the catalogue, the two invariants that make the difference between code that works and code that leaks goroutines in production.
First: whoever creates a channel is usually the one who closes it, and a channel is closed exactly once, by its single sender, after the last send. Closing from the receiver side or from multiple senders panics. Second: every goroutine must have a guaranteed exit. A goroutine blocked forever on a channel send that nobody will receive is a leak, and leaks accumulate silently until the process runs out of memory. The mechanism for guaranteeing an exit is a select that also watches a cancellation channel, almost always context.Context’s Done().
Keep both in mind as you read. Most of the patterns are variations on “start a goroutine, return the channel it feeds, and give it a way to stop.”
Generator: a function that returns a channel it feeds
A generator (or producer) is a function that starts a goroutine, returns a receive-only channel, and feeds that channel in the background. It turns “produce a sequence” into something the caller can range over, and it is the entry stage of almost every pipeline.
package main
import "fmt"
// generate emits each integer on its own channel, then closes it.
// The caller only ever sees a receive-only channel it can range over.
func generate(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
out <- n
}
}()
return out
}
func main() {
for n := range generate(2, 3, 5, 7, 11) {
fmt.Println(n)
}
}Output:
2
3
5
7
11The defer close(out) is what lets the caller’s range terminate: a range over a channel ends when the channel is closed and drained. The return type <-chan int is receive-only, so the compiler stops the caller from accidentally sending or closing.
Use when you want to decouple producing a sequence from consuming it, especially when items arrive over time (rows from a database cursor, lines from a file, events from a socket). Reach for a plain slice instead when the whole sequence is small and already in memory: a []int you range over has no goroutine, no channel, and no shutdown to reason about. A generator earns its cost only when the values are produced lazily or consumed concurrently.
Fan-out and fan-in: parallelize one stage, then merge
Fan-out means several goroutines read from the same input channel, spreading its work across cores. Fan-in means merging several channels back into one. Together they parallelize a slow stage without changing what the stages before and after it see.
package main
import (
"fmt"
"sort"
"sync"
)
func generate(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
out <- n
}
}()
return out
}
// square reads from in and emits n*n. Several of these run at once (fan-out),
// all pulling from the same input channel.
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
}
// merge fans in: it multiplexes several channels onto one output channel,
// closing it only after every input has drained.
func merge(channels ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
wg.Add(len(channels))
for _, c := range channels {
go func(c <-chan int) {
defer wg.Done()
for v := range c {
out <- v
}
}(c)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
func main() {
in := generate(1, 2, 3, 4, 5, 6, 7, 8)
// Fan out: three squarers share the one input channel.
worker1 := square(in)
worker2 := square(in)
worker3 := square(in)
var results []int
for v := range merge(worker1, worker2, worker3) {
results = append(results, v)
}
sort.Ints(results) // order is nondeterministic; sort for a stable print
fmt.Println(results)
}Output:
[1 4 9 16 25 36 49 64]Two details carry the pattern. Three square calls share the one in channel, so Go’s channel semantics distribute the values across them: each integer is received by exactly one squarer, whichever is free. And merge closes out only after a WaitGroup confirms every input goroutine has finished, which is the same close-after-Wait shape used everywhere in this catalogue. Results arrive in nondeterministic order, so the example sorts before printing.
Use when one stage is the bottleneck and its work is independent per item (image resizing, per-record HTTP enrichment, hashing). Reach for a single goroutine instead when the stage is already fast or the work is inherently sequential; fan-out only helps if the stage genuinely has parallel work to do. The dedicated fan-out fan-in guide covers bounding the fan-out width and handling errors from the merged stream.
Pipeline with per-stage cancellation
A pipeline is a chain of stages connected by channels: each stage receives from the previous one, does one transformation, and sends to the next. The failure mode is what happens when the consumer stops early. Without cancellation, an upstream stage blocks forever on a send nobody will receive, and every goroutine behind it leaks. Threading a context through every stage’s select fixes it.
package main
import (
"context"
"fmt"
)
func generate(ctx context.Context, nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
select {
case out <- n:
case <-ctx.Done(): // downstream stopped; abandon the rest
return
}
}
}()
return out
}
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
}
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
numbers := generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8)
squares := square(ctx, numbers)
// Consume only the first three, then cancel. Both upstream stages must
// notice ctx.Done() and exit instead of blocking forever on a send.
count := 0
for sq := range squares {
fmt.Println(sq)
count++
if count == 3 {
cancel()
break
}
}
fmt.Println("stopped early after", count)
}Output:
1
4
9
stopped early after 3Every stage sends inside a select that also watches ctx.Done(). When the consumer calls cancel() and breaks, the square goroutine’s next send unblocks via the ctx.Done() case and returns; the same happens up the chain to generate. No goroutine is left parked on a send. This is the single most important upgrade over the pipeline examples that omit context, because early consumer exit is the norm in real services (a client disconnects, a deadline fires, an error short-circuits the request). The fan-out and fan-in guide builds a multi-stage flow with bounded workers, errors, and measured throughput.
Use when data flows through distinct transformation steps and you want each step to run concurrently with the others. Reach for a plain function-call chain instead when the stages are cheap and synchronous: parse(validate(read(x))) needs no channels and cannot leak.
Worker pool: a fixed number of long-lived workers
A worker pool is fan-out with a fixed, deliberate number of workers pulling from a shared jobs channel. It is the pattern to use when work arrives continuously and you want to reuse workers rather than spawn one per job. Because it is the workhorse of production Go, it has its own full treatment.
The shape: N workers range over a jobs channel, send outcomes to a results channel, and a WaitGroup plus a closer goroutine handle shutdown. The single knob (worker count) is exactly how much concurrency you allow, which gives you backpressure and protects downstream services from a stampede.
Use when work is continuous or unbounded, when spawning a goroutine per item is itself too expensive, or when each worker should hold one scarce resource (a database connection, a file handle). Reach for a bare semaphore instead (next section) when you already have a finite slice in hand and just want to cap concurrency. The worker pool guide has the complete build: error propagation with a result struct, context cancellation, correct close ordering, and a measured worker-count curve for CPU-bound versus I/O-bound work.
Bounded concurrency with a semaphore channel
When you have a slice of items and want to process at most N at once, a buffered channel used as a counting semaphore is the least code. Each goroutine acquires a slot before starting and releases it when done; the buffer size is the concurrency limit.
package main
import (
"fmt"
"sync"
"sync/atomic"
"time"
)
func fetchPage(url string) int {
time.Sleep(30 * time.Millisecond) // simulate an HTTP round trip
return len(url)
}
func main() {
urls := make([]string, 40)
for i := range urls {
urls[i] = fmt.Sprintf("https://example.com/page/%d", i)
}
const limit = 8
sem := make(chan struct{}, limit) // buffered channel as a counting semaphore
var wg sync.WaitGroup
var totalBytes int64
start := time.Now()
for _, url := range urls {
sem <- struct{}{} // acquire a slot; blocks once 8 are in flight
wg.Add(1)
go func(url string) {
defer wg.Done()
defer func() { <-sem }() // release the slot
atomic.AddInt64(&totalBytes, int64(fetchPage(url)))
}(url)
}
wg.Wait()
fmt.Printf("fetched %d pages, cap %d, %d URL bytes, in %v\n",
len(urls), limit, totalBytes, time.Since(start).Round(10*time.Millisecond))
}Output:
fetched 40 pages, cap 8, 1070 URL bytes, in 150msThe sem <- struct{}{} before the go statement is the whole trick: once 8 slots are taken, the loop blocks there, so at most 8 fetches run concurrently. Forty pages at 30 ms, 8 at a time, is 5 batches, about 150 ms. Note struct{} for the element type, which allocates nothing, and sync/atomic for the shared counter because multiple goroutines write to it.
Use when the task list is finite, you do not need results streamed back through a channel, and you want the smallest possible amount of coordination code. Reach for golang.org/x/sync/semaphore instead when acquisition itself must respect a deadline (its Acquire takes a context) or when jobs have uneven cost and you want weighted slots. Reach for a full worker pool when work is continuous rather than a one-shot slice.
Future/promise: start work now, collect the value later
A future starts a computation immediately and hands back a channel that will eventually carry the single result. The caller can launch several futures, do other work, and only block when it actually needs each value. This is how you run independent lookups in parallel without a WaitGroup.
package main
import (
"fmt"
"time"
)
type Result struct {
Value int
Err error
}
// future starts the work immediately and hands back a channel that will
// deliver exactly one result. The buffer of 1 means the goroutine can finish
// and exit even if the caller never reads, so it cannot leak.
func future(userID int) <-chan Result {
ch := make(chan Result, 1)
go func() {
time.Sleep(50 * time.Millisecond) // simulate a slow lookup
ch <- Result{Value: userID * 10}
}()
return ch
}
func main() {
start := time.Now()
// Kick off both lookups; they run concurrently.
profile := future(1)
settings := future(2)
// ... do other useful work here while they run ...
p := <-profile // block only when we actually need the value
s := <-settings
fmt.Println("profile:", p.Value, "settings:", s.Value)
fmt.Println("elapsed:", time.Since(start).Round(10*time.Millisecond))
}Output:
profile: 10 settings: 20
elapsed: 50msBoth lookups run at once, so two 50 ms calls finish in 50 ms, not 100 ms. The buffer of 1 on the result channel is load-bearing: it lets the goroutine complete its send and exit even if the caller panics or returns early without reading. An unbuffered channel here would leak the goroutine in exactly that case. Carrying an error field in the result struct is what makes this production-shaped rather than a toy; real lookups fail, and the caller checks p.Err before using p.Value.
Use when you have a handful of independent async results to gather in one function and want them to read like ordinary values. Reach for errgroup or a worker pool instead when there are many of them or they share a cancellation policy; a channel per call does not scale to hundreds.
or-done and tee: safe ranging and splitting a stream
Two smaller utilities from the pipelines literature that solve recurring annoyances. orDone wraps a channel so ranging over it always respects cancellation, sparing every downstream loop from repeating the select. tee splits one input channel into two, delivering every value to both, which you need when two consumers must each see the full stream (for example, log every event and also aggregate it).
package main
import (
"context"
"fmt"
"sync"
)
// orDone wraps a channel so a range over it always respects cancellation.
// Without this, "for v := range in" ignores ctx entirely.
func orDone(ctx context.Context, in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for {
select {
case <-ctx.Done():
return
case v, ok := <-in:
if !ok {
return
}
select {
case out <- v:
case <-ctx.Done():
return
}
}
}
}()
return out
}
// tee splits one input channel into two. Every value is delivered to both
// outputs before the next value is read.
func tee(ctx context.Context, in <-chan int) (<-chan int, <-chan int) {
out1 := make(chan int)
out2 := make(chan int)
go func() {
defer close(out1)
defer close(out2)
for v := range orDone(ctx, in) {
out1, out2 := out1, out2 // shadow so we can nil them locally
for i := 0; i < 2; i++ {
select {
case <-ctx.Done():
case out1 <- v:
out1 = nil // sent to out1; disable this case
case out2 <- v:
out2 = nil // sent to out2; disable this case
}
}
}
}()
return out1, out2
}
func generate(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
out <- n
}
}()
return out
}
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
source := generate(1, 2, 3, 4, 5)
logs, sums := tee(ctx, source)
var wg sync.WaitGroup
wg.Add(2)
// Consumer 1: log every value.
go func() {
defer wg.Done()
for v := range logs {
fmt.Println("log:", v)
}
}()
// Consumer 2: total the values.
total := 0
go func() {
defer wg.Done()
for v := range sums {
total += v
}
}()
wg.Wait()
fmt.Println("sum:", total)
}Output:
log: 1
log: 2
log: 3
log: 4
log: 5
sum: 15The clever part of tee is the nil-channel trick. Setting out1 = nil after a successful send disables that select case (a send on a nil channel blocks forever), so the inner loop’s two iterations are guaranteed to send to each output exactly once, in whichever order they become ready. That is why both consumers see all five values. Setting a channel to nil to disable a select case is worth memorizing; it comes up whenever you want a select to “use up” a case. The mechanics of select, including the nil-channel idiom, are covered in Go select explained.
Use when two independent consumers each need the complete stream (tee), or when you have many downstream loops that would otherwise each reimplement cancellation (or-done). Reach for a plain range instead when there is one consumer and one context check; these helpers pay off at scale, not for a single loop.
Timeouts and cancellation with context and select
The most common concurrency requirement in a service is “do this, but give up after 100 ms or if the caller cancels.” The pattern is a select that races the real work against ctx.Done(). Whichever fires first wins, and the context carries both the timeout and the cancellation.
package main
import (
"context"
"errors"
"fmt"
"time"
)
// fetchWithTimeout runs a slow call in a goroutine and races it against the
// context. Whichever fires first through select wins.
func fetchWithTimeout(ctx context.Context) (string, error) {
result := make(chan string, 1) // buffered so the goroutine never blocks on send
go func() {
time.Sleep(200 * time.Millisecond) // downstream is slower than our deadline
result <- "payload"
}()
select {
case r := <-result:
return r, nil
case <-ctx.Done():
return "", ctx.Err()
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
start := time.Now()
_, err := fetchWithTimeout(ctx)
fmt.Println("err:", err)
fmt.Println("deadline exceeded:", errors.Is(err, context.DeadlineExceeded))
fmt.Println("returned after:", time.Since(start).Round(10*time.Millisecond))
}Output:
err: context deadline exceeded
deadline exceeded: true
returned after: 100msThe call returns after 100 ms with context.DeadlineExceeded, even though the underlying work needed 200 ms. Two things matter here. The result channel is buffered to 1, so when the timeout wins the race, the still-running goroutine can complete its send into the buffer and exit instead of blocking forever, which would leak it. And defer cancel() is not optional: a WithTimeout or WithCancel whose cancel is never called leaks the context’s internal timer goroutine until the deadline elapses. Inspect the reason with errors.Is(err, context.DeadlineExceeded) versus context.Canceled when you need to distinguish a timeout from an explicit cancel. The full range of context behavior lives in the context package guide.
Use when any operation crosses a boundary you do not fully control: a network call, a database query, anything that could hang. Reach for a plain call instead only for pure in-memory work that cannot block. In a real service, the context comes from the incoming request (r.Context() in an HTTP handler), so cancellation propagates automatically when the client disconnects.
Rate limiting with a ticker
To cap how fast you issue operations (respecting a downstream API’s requests-per-second limit), gate each operation on a time.Ticker. The ticker delivers a value on its channel at a fixed interval; receiving from it before each operation paces the loop exactly.
package main
import (
"fmt"
"time"
)
func main() {
requests := make(chan int, 5)
for i := 1; i <= 5; i++ {
requests <- i
}
close(requests)
// Allow one request every 50ms: a steady 20 requests/second.
limiter := time.NewTicker(50 * time.Millisecond)
defer limiter.Stop()
start := time.Now()
for req := range requests {
<-limiter.C // block until the next tick before proceeding
fmt.Printf("request %d sent at %v\n", req, time.Since(start).Round(10*time.Millisecond))
}
}Output:
request 1 sent at 50ms
request 2 sent at 100ms
request 3 sent at 150ms
request 4 sent at 200ms
request 5 sent at 250msEach <-limiter.C blocks until the next 50 ms tick, so the requests come out evenly spaced at 20 per second. defer limiter.Stop() releases the ticker’s resources; a ticker you forget to stop keeps its goroutine alive. This is the simplest limiter and it enforces a steady rate but no bursts. When you need to allow short bursts (a common real requirement, since traffic is bursty), a token-bucket limiter like golang.org/x/time/rate is the right tool. The Go middleware guide shows where request-wide controls belong in an HTTP stack.
Use when you must not exceed a fixed rate and a strictly even cadence is acceptable. Reach for a token bucket instead when bursts are allowed up to a cap, or when different callers need different rates.
Error propagation across goroutines
When you fan work out across goroutines, you need the first error to surface and, ideally, to cancel the rest. The standard-library pattern combines a WaitGroup, a buffered error channel, and a shared cancellable context. This is exactly what golang.org/x/sync/errgroup packages, but building it once shows what that library does.
package main
import (
"context"
"fmt"
"sync"
"time"
)
// runAll runs every task concurrently. The first task to fail cancels the
// shared context so the others can stop early, and Wait returns that error.
func runAll(ctx context.Context, tasks []func(context.Context) error) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
errCh := make(chan error, len(tasks))
var wg sync.WaitGroup
for _, task := range tasks {
wg.Add(1)
go func(task func(context.Context) error) {
defer wg.Done()
if err := task(ctx); err != nil {
select {
case errCh <- err:
cancel() // first error wins; signal the siblings
default: // buffer full: a later error we do not need
}
}
}(task)
}
wg.Wait()
close(errCh)
return <-errCh // first error, or nil if none were sent
}
func makeTask(id int, d time.Duration, fail bool) func(context.Context) error {
return func(ctx context.Context) error {
select {
case <-time.After(d):
if fail {
return fmt.Errorf("task %d failed", id)
}
return nil
case <-ctx.Done():
return ctx.Err()
}
}
}
func main() {
start := time.Now()
err := runAll(context.Background(), []func(context.Context) error{
makeTask(1, 200*time.Millisecond, false),
makeTask(2, 50*time.Millisecond, true), // fails first
makeTask(3, 500*time.Millisecond, false),
})
fmt.Println("err:", err)
fmt.Println("elapsed:", time.Since(start).Round(10*time.Millisecond))
}Output:
err: task 2 failed
elapsed: 50msTask 2 fails at 50 ms and calls cancel(). Tasks 1 and 3, which would have taken 200 ms and 500 ms, see ctx.Done() and return immediately, so the whole call finishes in about 50 ms instead of 500. The error channel is buffered to the task count so a failing goroutine never blocks on its send, and the select/default keeps only the first error while dropping the rest without blocking.
In production, most teams use errgroup.Group (verified against golang.org/x/sync v0.10.0) instead of hand-rolling this. errgroup.WithContext returns a group and a context; g.Go(func() error { ... }) starts a tracked goroutine; g.Wait() returns the first non-nil error and the context is cancelled the moment any goroutine fails. g.SetLimit(n) even turns it into a bounded pool. The code above is what errgroup does under the hood, which is why understanding it matters even if you never write it by hand.
Use when several concurrent operations must all succeed and one failure should abort the rest (fanning a request out to multiple backends). Reach for collecting every error instead (a []error joined with errors.Join) when it is a batch job that must attempt every item regardless of individual failures.
Graceful shutdown coordinating multiple goroutines
A long-running service must stop cleanly on SIGINT or SIGTERM: stop accepting new work, let in-flight work finish, and exit only once every goroutine has returned. The pattern is one cancellable context shared by all goroutines plus a WaitGroup that the main function blocks on.
package main
import (
"context"
"fmt"
"sync"
"sync/atomic"
"time"
)
func main() {
// In production this is signal.NotifyContext(context.Background(),
// os.Interrupt, syscall.SIGTERM); here we cancel on a timer to make the
// run reproducible.
ctx, cancel := context.WithCancel(context.Background())
jobs := make(chan int)
var processed int64
var wg sync.WaitGroup
// Three long-lived workers. Each finishes its current job before exiting.
for w := 0; w < 3; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case job := <-jobs:
time.Sleep(20 * time.Millisecond) // in-flight work we must not drop
_ = job
atomic.AddInt64(&processed, 1)
}
}
}()
}
// Producer feeds jobs until told to stop.
go func() {
for i := 0; ; i++ {
select {
case jobs <- i:
case <-ctx.Done():
return
}
}
}()
time.Sleep(200 * time.Millisecond)
fmt.Println("shutdown signal received")
cancel() // tell producer and workers to wind down
wg.Wait() // block until every worker has returned
fmt.Printf("processed %d jobs, all workers stopped cleanly\n", atomic.LoadInt64(&processed))
}Output:
shutdown signal received
processed 30 jobs, all workers stopped cleanlyThe shape is what matters: one cancel broadcasts to every goroutine at once (a closed channel makes every <-ctx.Done() ready), and wg.Wait() guarantees main does not exit until all three workers have returned from their current job. In a real service you replace the timer with signal.NotifyContext, which cancels the context on SIGINT or SIGTERM, and you would add a bounded shutdown deadline (context.WithTimeout on a second context) so a stuck worker cannot block shutdown forever. The job count is nondeterministic across runs (three workers, 20 ms each, 200 ms of runtime is around 30) but the clean-stop guarantee is not.
Use when you own the process lifecycle: an HTTP server, a queue consumer, any daemon. Reach for http.Server.Shutdown instead (or in addition) when the goroutines are HTTP handlers; the server’s own graceful-shutdown method drains connections for you.
Composing patterns: a small data-processing service
Real services chain several of these patterns. Here is a three-stage service in one file: a generator produces records, a bounded fan-out enriches them from a slow downstream, and a collector drains the results, all under a context deadline, with per-record errors carried through. It times itself so the payoff is concrete.
package main
import (
"context"
"fmt"
"sync"
"time"
)
type Record struct{ ID int }
type Enriched struct {
ID int
Payload string
Err error
}
// Stage 1: generator. Emits records, stops if the context is cancelled.
func generate(ctx context.Context, n int) <-chan Record {
out := make(chan Record)
go func() {
defer close(out)
for i := 1; i <= n; i++ {
select {
case out <- Record{ID: i}:
case <-ctx.Done():
return
}
}
}()
return out
}
// Stage 2: bounded fan-out. workers goroutines share the input channel and
// merge results onto one output channel.
func enrich(ctx context.Context, in <-chan Record, workers int) <-chan Enriched {
out := make(chan Enriched)
var wg sync.WaitGroup
wg.Add(workers)
for w := 0; w < workers; w++ {
go func() {
defer wg.Done()
for rec := range in {
select {
case <-time.After(30 * time.Millisecond): // simulate a downstream call
case <-ctx.Done():
return
}
res := Enriched{ID: rec.ID, Payload: fmt.Sprintf("data-%d", rec.ID)}
if rec.ID%20 == 0 {
res.Err = fmt.Errorf("record %d: enrichment failed", rec.ID)
}
select {
case out <- res:
case <-ctx.Done():
return
}
}
}()
}
go func() {
wg.Wait()
close(out)
}()
return out
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
const total = 100
start := time.Now()
// Stage 3: collector. Compose the stages and drain the final channel.
records := generate(ctx, total)
results := enrich(ctx, records, 10)
ok, failed := 0, 0
for e := range results {
if e.Err != nil {
failed++
continue
}
ok++
}
fmt.Printf("records: %d\n", total)
fmt.Printf("succeeded: %d\n", ok)
fmt.Printf("failed: %d\n", failed)
fmt.Printf("pooled (10 workers): %v\n", time.Since(start).Round(10*time.Millisecond))
fmt.Printf("sequential estimate: %v\n", time.Duration(total)*30*time.Millisecond)
}Output:
records: 100
succeeded: 95
failed: 5
pooled (10 workers): 310ms
sequential estimate: 3sOne hundred records through ten workers on 30 ms I/O work finishes in about 310 ms (ten batches of 30 ms) against a 3 second sequential estimate: a clean 10x. Every piece is a pattern from above. generate is the generator with cancellation. enrich is bounded fan-out (ten workers sharing one input) plus fan-in (all merging onto out, closed after wg.Wait()). The whole thing runs under a context deadline, and every send is inside a select on ctx.Done(), so if the deadline fired mid-run, all three stages would unwind without leaking a single goroutine. The five failures are records divisible by 20, carried through as Enriched.Err rather than crashing the pipeline. Swap time.After for a real HTTP or database call, pass the request’s context in, and this is shippable.
Common mistakes
Leaking goroutines when a stage exits early. The most frequent concurrency bug in Go. If a producer sends on an unbuffered channel and the consumer stops reading, the producer blocks on that send forever, and its goroutine (plus everything upstream) leaks:
// BROKEN: if the consumer breaks early, this goroutine blocks on send forever.
func generate() <-chan int {
out := make(chan int)
go func() {
for i := 0; ; i++ {
out <- i // no way out once the receiver stops
}
}()
return out
}The fix is the cancellation select used throughout this article: select { case out <- i: case <-ctx.Done(): return }. A leaked goroutine does not crash anything immediately, which is what makes it dangerous; it just holds memory until the process dies. The dedicated goroutine leaks guide shows how to detect them with runtime.NumGoroutine and go test -race.
Unbounded fan-out. Starting a goroutine per item with no limit is not a pattern, it is a resource exhaustion bug wearing a costume. Fifty thousand items become fifty thousand concurrent downstream calls that flatten your database or trip a rate limiter:
// BROKEN: nothing bounds the concurrency; 50,000 items = 50,000 live goroutines.
for _, item := range items {
go process(item)
}Bound it with a semaphore channel, a worker pool, or errgroup.SetLimit. The concurrency should always be a number you chose on purpose, not “however many items happened to arrive.”
Forgetting to propagate done/context. A pipeline where only some stages watch the context is as leaky as one with no context at all, because the stage that ignores it still blocks. Every stage that sends on a channel must include the ctx.Done() case in its select, and every blocking call inside a stage must take a context. Cancellation only works if it is threaded through the whole chain.
Reaching for channels where a mutex is simpler. Channels are for transferring ownership of data between goroutines; a mutex is for protecting shared state a few goroutines read and write. Guarding a single shared counter or map with a channel and a coordinating goroutine is more code, slower, and harder to read than a sync.Mutex:
// Overcomplicated: a whole goroutine and channel to guard one integer.
// A sync.Mutex around the counter is fewer lines and faster.The rule of thumb from the Go proverbs: share memory by communicating (channels) when you are passing work along; but when several goroutines just touch the same variable, a mutex is the frank tool. The sync.Mutex guide covers exactly when each wins.
What next
You now have a catalogue you can lift code from, plus the judgment to pick the pattern that fits and skip the one that does not. Go deeper on the pieces:
- Fan-out fan-in in depth expands the stage-based pattern with bounding, error handling, and measured throughput.
- The worker pool guide is the full production build of the pattern this article summarized, including how to size the pool with a benchmark.
- Goroutine leaks and the context package guide cover the two things that separate correct concurrent code from code that quietly rots in production.
- Concurrency dominates Go interviews; the Go interview questions collection has a full concurrency section with worked answers.
Run each example with -race on, then change a worker count or a timeout and watch the output move. The patterns stick faster when you break them and fix them yourself.