Go Worker Pool Pattern: Examples, Channels, and Shutdown

A runnable Go worker pool#

Save this as main.go. Three workers process five jobs through unbuffered channels:

package main

import (
	"fmt"
	"sync"
)

type Result struct {
	Job   int
	Value int
}

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

func main() {
	const workers = 3
	jobs := make(chan int)
	results := make(chan Result)

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

	go func() {
		defer close(jobs)
		for job := 1; job <= 5; job++ {
			jobs <- job
		}
	}()

	go func() {
		wg.Wait()
		close(results)
	}()

	for result := range results {
		fmt.Printf("job %d: %d\n", result.Job, result.Value)
	}
}

Run go run main.go. You get one square for each job, with no guarantee about print order. Job 1 produces 1, job 2 produces 4, and so on through job 5 producing 25. A worker that finishes first can send first, regardless of submission order.

The producer runs in a separate goroutine because main must receive results while workers execute. If main sent every job before reading results, a worker could block sending its result while main blocked sending the next job. Giving both channels enough buffer space for the whole batch would hide that dependency and make memory use grow with batch size.

Here, each unbuffered send waits for a receiver. A bounded buffer can absorb a short burst, but once it fills, the producer still waits. That waiting supplies backpressure. Choose the queue capacity from the amount of waiting work you can afford to retain.

Who closes each channel?#

The producer owns jobs and closes it after its last send. Workers finish their range loops after they consume all submitted jobs. Each worker calls Done, and the coordinator closes results after Wait returns. The collector then leaves its own range loop.

A worker cannot close the shared results channel: another worker might still send to it. The collector cannot close it to tell workers to stop either; sending on a closed channel panics. Use a separate cancellation signal for early termination.

Call wg.Add(workers) before launching the goroutines. This registers the work before the coordinator can call Wait. The Go pipeline article describes the same channel ownership and fan-in rules.

This example drains the batch during normal shutdown. Once submitted, a job runs to completion and produces one result. It has no timeout, and the collector must keep reading until results closes.

Cancellation when the caller stops reading#

An HTTP client may disconnect or a batch deadline may expire. If the collector returns while a worker is sending, the first example leaves that worker blocked. A cancellation-aware pool needs an exit path at both channel boundaries and inside any long-running job.

Save this separate example as pool.go in a new directory. Squares returns a results channel and a done channel. Closing done confirms that the producer and all workers have finished.

package pool

import (
	"context"
	"sync"
)

type Result struct {
	Job   int
	Value int
}

func Squares(ctx context.Context, workers int, input []int) (<-chan Result, <-chan struct{}) {
	if workers < 1 {
		panic("workers must be positive")
	}
	jobs := make(chan int)
	results := make(chan Result)
	done := make(chan struct{})

	var wg sync.WaitGroup
	wg.Add(workers + 1)
	go func() {
		defer wg.Done()
		defer close(jobs)
		for _, job := range input {
			select {
			case <-ctx.Done():
				return
			case jobs <- job:
			}
		}
	}()

	for i := 0; i < workers; i++ {
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case job, ok := <-jobs:
					if !ok {
						return
					}
					result := Result{Job: job, Value: job * job}
					select {
					case <-ctx.Done():
						return
					case results <- result:
					}
				}
			}
		}()
	}

	go func() {
		wg.Wait()
		close(results)
		close(done)
	}()
	return results, done
}

Keep input unchanged until done closes: the producer reads the supplied slice without copying it. The function rejects zero workers because no goroutine would consume jobs. For a public API that accepts user configuration, returning a validation error may fit better than a panic.

If your caller stops collecting, call its cancel function and wait for <-done before releasing resources the workers use. Cancellation may discard jobs and results. Use the first example’s drain behavior when every accepted job must finish; use cancellation when incomplete work is acceptable.

A select does not give cancellation priority over another ready case. A worker may receive or publish a value after cancellation becomes ready. This pool promises eventual termination for its finite arithmetic jobs, not a strict “no more results after cancel” boundary.

For an HTTP request or database query, pass ctx into the context-aware operation too. A cancellable channel send cannot interrupt a function that ignores cancellation. The context cancellation guide covers how that signal reaches the work itself.

Test completion and an abandoned collector#

Put this in pool_test.go beside pool.go. The first test checks the results without relying on worker order. The second deliberately leaves results unread: with unbuffered results, workers cannot publish, so cancellation must release them and allow done to close.

package pool

import (
	"context"
	"testing"
	"time"
)

func TestSquares(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	results, done := Squares(ctx, 3, []int{1, 2, 3, 4, 5})
	seen := make(map[int]int)
	for result := range results {
		seen[result.Job]++
		if result.Value != result.Job*result.Job {
			t.Fatalf("wrong result: %+v", result)
		}
	}
	<-done
	for job := 1; job <= 5; job++ {
		if seen[job] != 1 {
			t.Fatalf("job %d appeared %d times", job, seen[job])
		}
	}
}

func TestCancelWithoutReadingResults(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	_, done := Squares(ctx, 2, []int{1, 2, 3, 4, 5})
	cancel()
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("pool did not stop after cancellation")
	}
}

In that directory, run:

go mod init example.com/workerpool
go test -race -count=20 ./...

The timeout bounds a failure; it does not assume a scheduling order or wait an arbitrary sleep before cancellation. The test covers an abandoned collector and shutdown, though it does not force every possible scheduler interleaving. The race detector checks the memory accesses exercised by these test runs. Test new job implementations under the race detector too.

For a pool that calls external services, add tests for job errors, cancellation during a call, and the maximum number of simultaneous calls. Test these contracts before asserting how many goroutines happen to exist in the process. The testing article explains that distinction.

Choose worker count from the constrained resource#

For CPU-heavy work, start near the runtime’s available parallelism, inspect runtime.GOMAXPROCS(0), and measure throughput and latency with representative input. More workers can add contention without increasing completed work. There is no universal reason to set GOMAXPROCS or use 2 * runtime.NumCPU() just because you created a pool.

For database work, reserve connection capacity for other requests. A pool of 50 workers cannot make 10 available connections execute 50 queries at once. If each job holds one connection, pick a worker budget that fits the connections you can dedicate to this batch.

For an external API, concurrent requests and requests per second are separate limits. A worker pool caps in-flight work. Fast requests can still exceed a rate quota, so add rate limiting when the service imposes one. Measure queue wait as well as execution time: a short handler can still serve a slow request if it waits behind a large batch.

Bound memory beyond the goroutine count. The cancellable example retains its input slice until the producer finishes. For a large file or endless stream, read and submit incrementally through a bounded queue instead of loading all jobs first. Result collection can also grow without bound if the caller appends every result to a slice.

Results, errors, and ordering#

The arithmetic examples use small integers and cannot report a job error. Real jobs need an explicit error policy. Add a job identifier and an Err error field to a result type if the collector should record failures and continue. Cancel the shared context if one failure makes the remaining work useless. Retrying inside a worker consumes the same concurrency slot; cap attempts and make side effects safe to repeat before adding retries.

If input order matters, attach an index and write each result into its assigned slot, or let the collector reorder results. A slow early job can hold later results in memory. Streaming in completion order avoids that waiting when the consumer does not need ordering.

A panic in a worker terminates the program unless that goroutine recovers it. Decide whether recovery belongs at your application boundary; converting every panic into a routine job error can hide a programming defect. Neither example above attempts panic recovery or persistent delivery across process restarts.

Worker pool or semaphore?#

Choose a worker pool when a queue and a fixed set of consumers match the job lifecycle. You can observe pending work, decide when to stop accepting jobs, and drain the queue before shutdown.

A semaphore fits when callers already own each task and need a shared concurrency limit around an operation. Acquire before launching a goroutine if the goroutine count itself must remain bounded. Launching thousands of goroutines that then wait on a semaphore still allocates thousands of goroutines. The semaphore pattern in Go walks through that alternative.

Before using this pool for real work, replace the square operation with one representative job. Set its concurrency budget from a connection limit, CPU capacity, or service quota, then test both a full drain and a caller that cancels without reading another result.