Go · lesson 16 of 25
Concurrency Patterns in Go
Go concurrency patterns run for real: worker pools, bounded fan-out pipelines, context timeouts and cancellation, RWMutex, sync.Once, and errgroup limits.
By SkillByExample · Updated
The goroutines and channels lessons cover the primitives. This one assembles them into the patterns real services use. Concurrent code finishes in a different order on each run, so every example below either writes results into indexed slots or sorts them before printing; each was run repeatedly and with GOMAXPROCS=1 to confirm the output is identical every time.
Worker pool
A fixed number of workers read from a shared jobs channel. Closing jobs tells them there is no more work.
package main
import (
"fmt"
"sync"
)
type result struct {
id, out int
}
func worker(jobs <-chan int, results chan<- result, wg *sync.WaitGroup) {
defer wg.Done()
for n := range jobs { // ends when jobs is closed and drained
results <- result{n, n * n}
}
}
func main() {
const workers, jobCount = 3, 8
jobs := make(chan int)
results := make(chan result)
var wg sync.WaitGroup
for range workers {
wg.Add(1)
go worker(jobs, results, &wg)
}
go func() {
for i := range jobCount {
jobs <- i
}
close(jobs)
}()
go func() {
wg.Wait()
close(results) // only after every worker has stopped sending
}()
out := make([]int, jobCount)
for r := range results {
out[r.id] = r.out
}
fmt.Println(out)
}$ go run pool.go
[0 1 4 9 16 25 36 49]Three goroutines did eight jobs. The results arrived in an unpredictable order, so each carries its job id and lands in its own slot. The goroutine that waits and then closes results is the important detail: results may only be closed once no worker can send any more, and main can’t wait itself because it is busy receiving.
Bounded fan-out with errgroup
golang.org/x/sync/errgroup replaces the hand-built WaitGroup + error plumbing. SetLimit caps concurrency, and WithContext cancels the others when one fails.
package main
import (
"context"
"fmt"
"sync/atomic"
"golang.org/x/sync/errgroup"
)
func main() {
var running, peak atomic.Int32
sizes := make([]int, 10)
g := new(errgroup.Group)
g.SetLimit(3) // at most 3 goroutines at a time
for i := range sizes {
g.Go(func() error {
n := running.Add(1)
defer running.Add(-1)
for p := peak.Load(); n > p && !peak.CompareAndSwap(p, n); p = peak.Load() {
}
sizes[i] = i * 10
return nil
})
}
fmt.Println(g.Wait(), sizes, "peak<=3:", peak.Load() <= 3)
g2, ctx := errgroup.WithContext(context.Background())
g2.Go(func() error { return fmt.Errorf("shard 2: disk full") })
g2.Go(func() error {
<-ctx.Done() // cancelled when the first goroutine fails
return ctx.Err()
})
fmt.Println(g2.Wait())
}$ go get golang.org/x/sync/errgroup@v0.19.0
go: added golang.org/x/sync v0.19.0
$ go run .
<nil> [0 10 20 30 40 50 60 70 80 90] peak<=3: true
shard 2: disk fullThe version is pinned to v0.19.0 because it’s the newest x/sync release whose go.mod accepts Go 1.24, the toolchain used here; asking for a later one makes go get raise your module’s go line and download a newer toolchain to match. g.Go blocks once three goroutines are running, so at most three were ever active — the peak counter checks it. In the second group, the first error cancelled ctx, which released the other goroutine; Wait returned the first error, not context canceled. That’s the shape for “fetch these N things, give up on the first failure”.
Pipeline with parallel stages
Stages connected by channels, with the middle stage fanned out across several goroutines.
package main
import (
"fmt"
"slices"
"strings"
"sync"
)
func source(words ...string) <-chan string {
out := make(chan string)
go func() {
defer close(out)
for _, w := range words {
out <- w
}
}()
return out
}
func upper(in <-chan string, workers int) <-chan string {
out := make(chan string)
var wg sync.WaitGroup
for range workers {
wg.Add(1)
go func() {
defer wg.Done()
for w := range in {
out <- strings.ToUpper(w)
}
}()
}
go func() { wg.Wait(); close(out) }()
return out
}
func main() {
var got []string
for w := range upper(source("go", "is", "fun", "and", "fast"), 3) {
got = append(got, w)
}
slices.Sort(got)
fmt.Println(got)
}$ go run pipeline.go
[AND FAST FUN GO IS]Fanning a stage out loses ordering: three goroutines race to send, so the collected slice is sorted before printing. If order matters, carry an index with each item as the worker pool does. Each stage owns and closes its own output, so shutdown flows down the pipeline when the source closes.
Context: timeouts
context.WithTimeout gives a context whose Done channel closes after the deadline. Functions that take a ctx should give up when it does.
package main
import (
"context"
"errors"
"fmt"
"time"
)
func query(ctx context.Context, ready <-chan string) (string, error) {
select {
case r := <-ready:
return r, nil
case <-ctx.Done():
return "", fmt.Errorf("query: %w", ctx.Err())
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel() // always release the context's timer
fast := make(chan string, 1)
fast <- "3 rows"
fmt.Println(query(ctx, fast))
never := make(chan string)
_, err := query(ctx, never)
fmt.Println(err)
fmt.Println(errors.Is(err, context.DeadlineExceeded))
}$ go run ctxtimeout.go
3 rows <nil>
query: context deadline exceeded
trueThe second query waited until the 50 ms deadline and returned a wrapped context.DeadlineExceeded. defer cancel() is required even when the timeout fires on its own: it releases the timer early on the success path, and go vet warns when a cancel function is dropped.
Context: cancellation down a call chain
Cancelling a parent context cancels every context derived from it.
package main
import (
"context"
"fmt"
"sync"
)
func count(ctx context.Context, name string, out chan<- string) {
for i := 0; ; i++ {
select {
case out <- fmt.Sprint(name, i):
case <-ctx.Done():
return
}
}
}
func main() {
ctx, cancel := context.WithCancel(context.Background())
out := make(chan string)
var wg sync.WaitGroup
for _, name := range []string{"a", "b", "c"} {
wg.Add(1)
go func() { defer wg.Done(); count(ctx, name, out) }()
}
got := 0
for range out {
if got++; got == 10 {
break
}
}
cancel()
wg.Wait() // every goroutine has returned
fmt.Println("received", got, "values; workers stopped:", ctx.Err())
}$ go run ctxcancel.go
received 10 values; workers stopped: context canceledThree producers would run forever. After ten values, one cancel() closed ctx.Done() for all of them, and wg.Wait() proves they all returned — no leaked goroutines. A context.Context is the first parameter of any function that may block, by convention named ctx.
RWMutex for read-heavy state
sync.RWMutex lets any number of readers in at once, but a writer gets exclusive access.
package main
import (
"fmt"
"sync"
)
type Cache struct {
mu sync.RWMutex
data map[string]string
}
func (c *Cache) Get(k string) (string, bool) {
c.mu.RLock()
defer c.mu.RUnlock()
v, ok := c.data[k]
return v, ok
}
func (c *Cache) Set(k, v string) {
c.mu.Lock()
defer c.mu.Unlock()
c.data[k] = v
}
func main() {
c := &Cache{data: map[string]string{}}
var wg sync.WaitGroup
for i := range 50 {
wg.Add(2)
go func() { defer wg.Done(); c.Set(fmt.Sprint("k", i%5), "v") }()
go func() { defer wg.Done(); c.Get("k0") }()
}
wg.Wait()
fmt.Println(len(c.data))
}$ go run -race rwmutex.go
5100 concurrent goroutines, no race reported, and five distinct keys. RLock is only worth it when reads dominate and hold the lock for a while; for tiny critical sections a plain Mutex is often as fast. Either way, the mutex lives next to the data it guards, and the struct must not be copied once used — go vet flags copies of a value containing a mutex.
sync.Once
Once.Do(f) runs f exactly once, however many goroutines call it, and the others wait until it has finished.
package main
import (
"fmt"
"sync"
)
var (
loadOnce sync.Once
config map[string]string
loads int
)
func Config() map[string]string {
loadOnce.Do(func() {
loads++
config = map[string]string{"env": "prod"}
})
return config
}
func main() {
var wg sync.WaitGroup
for range 20 {
wg.Add(1)
go func() { defer wg.Done(); _ = Config()["env"] }()
}
wg.Wait()
fmt.Println("loads:", loads, "env:", Config()["env"])
}$ go run -race once.go
loads: 1 env: prodTwenty goroutines raced to call Config, the loader ran once, and the race detector is satisfied: Once guarantees that everything f wrote is visible to every caller after Do returns. Go 1.21 added sync.OnceValue for the common case of computing a single value: var Config = sync.OnceValue(loadConfig).
Practice
1. Change pool.go so each worker also returns an error for negative inputs, carried in result, and feed it the jobs 3, -1, 4.
Solution
package main
import (
"fmt"
"sync"
)
type result struct {
id, out int
err error
}
func main() {
inputs := []int{3, -1, 4}
jobs, results := make(chan int), make(chan result)
var wg sync.WaitGroup
for range 2 {
wg.Add(1)
go func() {
defer wg.Done()
for id := range jobs {
n := inputs[id]
if n < 0 {
results <- result{id: id, err: fmt.Errorf("job %d: negative input %d", id, n)}
continue
}
results <- result{id: id, out: n * n}
}
}()
}
go func() {
for id := range inputs {
jobs <- id
}
close(jobs)
}()
go func() { wg.Wait(); close(results) }()
out := make([]result, len(inputs))
for r := range results {
out[r.id] = r
}
for _, r := range out {
fmt.Println(r.out, r.err)
}
}$ go run p1.go
9 <nil>
0 job 1: negative input -1
16 <nil>The error travels with the result instead of stopping the pool, so one bad job doesn’t lose the others.
2. Use context.WithTimeout to give a loop that does one step every 20 ms a budget of 70 ms, and report how many steps it completed.
Solution
package main
import (
"context"
"fmt"
"time"
)
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 70*time.Millisecond)
defer cancel()
tick := time.NewTicker(20 * time.Millisecond)
defer tick.Stop()
steps := 0
for {
select {
case <-tick.C:
steps++
case <-ctx.Done():
fmt.Println("steps:", steps, "err:", ctx.Err())
return
}
}
}$ go run p2.go
steps: 3 err: context deadline exceededTicks at 20, 40 and 60 ms fit inside the budget; the deadline fires at 70 ms before the fourth. On a heavily loaded machine a tick can be delayed, so treat the exact count as typical rather than guaranteed.
3. Rewrite the loader in once.go with sync.OnceValue.
Solution
package main
import (
"fmt"
"sync"
)
var loads int
var Config = sync.OnceValue(func() map[string]string {
loads++
return map[string]string{"env": "prod"}
})
func main() {
var wg sync.WaitGroup
for range 20 {
wg.Add(1)
go func() { defer wg.Done(); _ = Config()["env"] }()
}
wg.Wait()
fmt.Println("loads:", loads, "env:", Config()["env"])
}$ go run -race p3.go
loads: 1 env: prodOnceValue returns a function that computes the value on first call and returns the cached result after that — no separate Once and package variable needed.
4. Add a third stage to pipeline.go that keeps only words longer than 2 characters, without changing the other stages.
Solution
package main
import (
"fmt"
"slices"
"strings"
)
func source(words ...string) <-chan string {
out := make(chan string)
go func() {
defer close(out)
for _, w := range words {
out <- w
}
}()
return out
}
func filter(in <-chan string, keep func(string) bool) <-chan string {
out := make(chan string)
go func() {
defer close(out)
for w := range in {
if keep(w) {
out <- w
}
}
}()
return out
}
func main() {
long := filter(source("go", "is", "fun", "and", "fast"), func(w string) bool { return len(w) > 2 })
var got []string
for w := range long {
got = append(got, strings.ToUpper(w))
}
slices.Sort(got)
fmt.Println(got)
}$ go run p4.go
[AND FAST FUN]A stage is just a function from one channel to another, so stages compose like ordinary function calls.