A worker pool is a fixed set of goroutines that pull work from a shared queue. You create N workers at startup and feed them jobs — rather than spawning a new goroutine for every unit of work. This pattern gives you controlled concurrency: N determines the maximum number of things happening simultaneously, which protects downstream resources (databases, APIs, file handles) from being overwhelmed.
Go’s channels make the pattern natural. The job channel is the queue; closing it signals workers to stop; a WaitGroup or done channel signals when they’re all finished.
For foundations see Go goroutines, Go channels, and Go semaphores.
The Core Worker Pool
The minimal complete pattern: a job channel, a results channel, N workers, and clean shutdown:
type Job struct {
ID int
Data string
}
type Result struct {
JobID int
Output string
Err error
}
func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs { // range exits cleanly when jobs is closed
out, err := process(job.Data)
results <- Result{JobID: job.ID, Output: out, Err: err}
}
}
func runPool(jobs []Job, numWorkers int) []Result {
jobCh := make(chan Job, numWorkers)
resultCh := make(chan Result, len(jobs))
var wg sync.WaitGroup
for i := 0; i < numWorkers; i++ {
wg.Add(1)
go worker(i, jobCh, resultCh, &wg)
}
// Send all jobs, then close to signal workers to stop
for _, j := range jobs {
jobCh <- j
}
close(jobCh)
// Wait for workers, then close results so the collector can finish
go func() {
wg.Wait()
close(resultCh)
}()
var results []Result
for r := range resultCh {
results = append(results, r)
}
return results
}
range jobs is the idiomatic way to drain a channel — workers loop until the channel is closed, then exit naturally, triggering wg.Done(). The goroutine that calls wg.Wait() and then close(resultCh) is a standard pattern to signal the result collector when all work is done.
Sizing the pool: for I/O-bound work (HTTP calls, DB queries), use 10–50 workers — they spend most of their time waiting, so more workers means more in-flight requests without burning CPU. For CPU-bound work (encoding, hashing), use runtime.NumCPU() — adding more workers than CPUs just adds context-switching overhead.
Adding Context for Cancellation
Production pools need to respect cancellation — a timeout, a client disconnect, or a SIGTERM should stop all workers promptly:
func worker(ctx context.Context, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case job, ok := <-jobs:
if !ok {
return // channel closed, exit
}
out, err := processWithContext(ctx, job)
select {
case results <- Result{JobID: job.ID, Output: out, Err: err}:
case <-ctx.Done():
return
}
case <-ctx.Done():
return
}
}
}
The select on ctx.Done() in both the job-receive and result-send positions means workers stop immediately on cancellation, even if the jobs channel still has pending work or the results channel is full.
Fan-Out / Fan-In
Fan-out distributes work from one source to multiple workers. Fan-in merges results from multiple goroutines into one channel. Together they form the backbone of parallel processing pipelines.
The key to fan-in is a dedicated merge goroutine per input channel, all writing to a shared output:
// fanIn merges multiple result channels into one
func fanIn(ctx context.Context, channels ...<-chan Result) <-chan Result {
out := make(chan Result, len(channels))
var wg sync.WaitGroup
forward := func(ch <-chan Result) {
defer wg.Done()
for r := range ch {
select {
case out <- r:
case <-ctx.Done():
return
}
}
}
wg.Add(len(channels))
for _, ch := range channels {
go forward(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
Fan-out is just spawning multiple workers that all read from the same job channel — the pattern shown in the core pool above. Fan-in is useful when you have separate worker groups producing results on different channels and want to unify them for consumption.
Pipeline Pattern
A pipeline chains stages where each stage reads from the previous stage’s output channel. Each stage runs concurrently — while stage 2 is processing item N, stage 1 is already working on item N+1.
// Stage 1: generate work
func generate(ctx context.Context, items []string) <-chan string {
out := make(chan string)
go func() {
defer close(out)
for _, s := range items {
select {
case out <- s:
case <-ctx.Done():
return
}
}
}()
return out
}
// Stage 2: transform
func transform(ctx context.Context, in <-chan string) <-chan string {
out := make(chan string)
go func() {
defer close(out)
for s := range in {
select {
case out <- strings.ToUpper(s):
case <-ctx.Done():
return
}
}
}()
return out
}
// Stage 3: filter
func filter(ctx context.Context, in <-chan string, pred func(string) bool) <-chan string {
out := make(chan string)
go func() {
defer close(out)
for s := range in {
if pred(s) {
select {
case out <- s:
case <-ctx.Done():
return
}
}
}
}()
return out
}
// Wire stages together
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
items := []string{"hello", "world", "foo", "bar", "go"}
stage1 := generate(ctx, items)
stage2 := transform(ctx, stage1)
stage3 := filter(ctx, stage2, func(s string) bool { return len(s) > 3 })
for result := range stage3 {
fmt.Println(result) // HELLO, WORLD
}
}
Each stage function accepts a channel and returns a channel, making them composable. The context threads through every stage so cancellation propagates immediately to all running goroutines.
Backpressure
Backpressure means slowing the producer when consumers can’t keep up — rather than letting the queue grow unboundedly and exhaust memory. In Go, a bounded channel provides backpressure automatically:
// If jobCh is full, the sender blocks until a worker consumes a job
jobCh := make(chan Job, 100) // max 100 jobs queued at any time
// Producer blocks here when the channel is full
for _, job := range allJobs {
jobCh <- job // natural backpressure
}
The buffer size determines the tradeoff between throughput and memory. A buffer of numWorkers lets all workers stay busy without queuing much; a larger buffer smooths out bursts but uses more memory.
For more explicit backpressure — where you want to measure and act on queue depth:
func submitWithBackpressure(ctx context.Context, jobCh chan<- Job, job Job) error {
select {
case jobCh <- job:
return nil
case <-ctx.Done():
return ctx.Err()
case <-time.After(100 * time.Millisecond):
return fmt.Errorf("queue full, retry later")
}
}
This pattern is common in HTTP handlers that feed work to a background pool — if the pool is saturated, the handler returns 503 instead of queuing indefinitely.
Error Propagation
One goroutine failing should stop the rest. errgroup from golang.org/x/sync handles this elegantly:
import "golang.org/x/sync/errgroup"
func processAll(ctx context.Context, items []string) error {
g, ctx := errgroup.WithContext(ctx) // derived context is cancelled on first error
jobs := make(chan string, len(items))
for _, item := range items {
jobs <- item
}
close(jobs)
for i := 0; i < 5; i++ {
g.Go(func() error {
for job := range jobs {
if err := processOne(ctx, job); err != nil {
return err // cancels ctx, unblocks other goroutines
}
}
return nil
})
}
return g.Wait() // returns first non-nil error
}
errgroup.WithContext returns a derived context that’s cancelled the moment any goroutine in the group returns a non-nil error. All other goroutines see ctx.Done() close and can exit their current work early. g.Wait() collects and returns the first error.
When to Use a Semaphore Instead
A worker pool requires pre-allocation — you decide N at pool creation time. A semaphore (buffered channel) lets goroutines spawn freely but gates them at a checkpoint. Choose based on your creation cost:
- Pre-created goroutines are cheaper: use a worker pool when goroutines need to maintain expensive state (a database connection, an open file, a warm HTTP client).
- Per-request goroutines are fine: use a semaphore when each goroutine is stateless and the main concern is just bounding concurrency. See Go semaphores for the full pattern.
Summary
- Worker pools: fixed N goroutines drain a job channel;
close(jobs)shuts them down cleanly; aWaitGroupsignals completion - Size CPU-bound pools at
runtime.NumCPU(); size I/O-bound pools based on downstream capacity (start at 10–20, tune with load tests) - Thread a
context.Contextthrough every stage and everyselect— cancellation must reach all goroutines - Fan-in with one forwarding goroutine per input channel, merged into a shared output
- Use bounded job channels for natural backpressure; return an error from the producer when the queue is full rather than blocking indefinitely
- Use
errgroupwhen the first failure should cancel all other work
Resources
- Go Blog: Pipelines and Cancellation
- golang.org/x/sync/errgroup
- Go by Example: Worker Pools
- Concurrency in Go (book) — Katherine Cox-Buday
Comments