Concurrency Patterns in Go

Worker Pool

Launching one goroutine per task works well for small numbers, but unbounded concurrency causes problems: too many parallel HTTP requests gets you rate-limited, too many parallel DB queries exhausts your connection pool, and too many CPU-bound goroutines causes context-switching overhead. A worker pool solves this by keeping a fixed number of goroutines running — they all read from a shared job channel, so work is distributed automatically without spinning up new goroutines for every task.

package main

import (
    "fmt"
    "sync"
    "time"
)

type Job struct {
    ID    int
    Input int
}

type Result struct {
    Job    Job
    Output int
}

// worker reads jobs from the jobs channel until it is closed, then exits
func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        // simulate work (e.g. a DB query or CPU computation)
        time.Sleep(10 * time.Millisecond)
        results <- Result{
            Job:    job,
            Output: job.Input * job.Input, // square the input
        }
        fmt.Printf("worker %d: job %d → %d\n", id, job.ID, job.Input*job.Input)
    }
}

func main() {
    const numWorkers = 3
    const numJobs = 10

    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)

    // Start the fixed pool of workers
    var wg sync.WaitGroup
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // Send all jobs, then close the channel to signal workers to stop
    for i := 1; i <= numJobs; i++ {
        jobs <- Job{ID: i, Input: i}
    }
    close(jobs) // workers exit their range loop when jobs is closed

    // Close results once all workers are done — in a separate goroutine
    // because we need to drain results below and wg.Wait would deadlock otherwise
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    for r := range results {
        _ = r
    }
    fmt.Println("all jobs done")
}

Pipeline Pattern

Pipelines are a natural fit for multi-stage data processing where each stage transforms its input and passes the result to the next. Because each stage runs in its own goroutine and stages communicate via channels, all stages run concurrently — stage 2 processes item 1 while stage 1 is already working on item 2. This streaming model reduces both latency and peak memory usage compared to batch processing.

// generate emits a sequence of numbers onto a channel
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out) // signal: no more numbers
    }()
    return out
}

// square receives numbers and emits their squares
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

// filter passes through only values that satisfy the predicate
func filter(in <-chan int, pred func(int) bool) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            if pred(n) {
                out <- n
            }
        }
        close(out)
    }()
    return out
}

func main() {
    nums := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    squared := square(nums)
    evens := filter(squared, func(n int) bool { return n%2 == 0 })

    for v := range evens {
        fmt.Println(v) // 4, 16, 36, 64, 100
    }
}

Context Cancellation

context.Context is Go’s standard mechanism for propagating cancellation signals and deadlines through a call chain. Every goroutine that does I/O or blocks should accept a context and check ctx.Done(). When a timeout fires or a caller cancels, all goroutines in the chain exit cleanly — preventing the goroutine leaks that would otherwise accumulate in a long-running service. Always defer cancel() immediately after creating a context with a deadline or cancellation.

import (
    "context"
    "fmt"
    "time"
)

func doWork(ctx context.Context, id int) error {
    for i := 0; ; i++ {
        select {
        case <-ctx.Done():
            // ctx.Err() returns context.DeadlineExceeded or context.Canceled
            fmt.Printf("worker %d: cancelled after %d iterations\n", id, i)
            return ctx.Err()
        default:
            // do a unit of work
            time.Sleep(50 * time.Millisecond)
            fmt.Printf("worker %d: iteration %d\n", id, i)
        }
    }
}

func main() {
    // Automatically cancelled after 200ms
    ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
    defer cancel() // always defer — frees resources even if timeout fires first

    if err := doWork(ctx, 1); err != nil {
        fmt.Println("error:", err) // error: context deadline exceeded
    }
}

Context with cancel for manual cancellation

func main() {
    ctx, cancel := context.WithCancel(context.Background())

    var wg sync.WaitGroup
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            doWork(ctx, id)
        }(i)
    }

    time.Sleep(150 * time.Millisecond)
    cancel() // broadcast stop signal to all three workers at once

    wg.Wait()
    fmt.Println("all workers stopped")
}

Passing context through a call chain

// ctx flows from the HTTP handler down through every layer —
// if the client disconnects, all downstream operations are cancelled automatically
func handleRequest(ctx context.Context, userID int) error {
    user, err := fetchUser(ctx, userID)
    if err != nil {
        return fmt.Errorf("handleRequest: %w", err)
    }
    return sendEmail(ctx, user.Email)
}

func fetchUser(ctx context.Context, id int) (User, error) {
    // db.QueryRowContext respects ctx — cancelled if the client disconnects
    row := db.QueryRowContext(ctx, "SELECT id, email FROM users WHERE id = $1", id)
    // ...
}

sync.Mutex — Protecting Shared State

A mutex (mutual exclusion lock) protects a block of code so only one goroutine can execute it at a time. Use defer mu.Unlock() immediately after Lock() — it guarantees the unlock happens even if the function panics or returns early. Embed the mutex in the struct it protects and never copy a struct containing a mutex after first use.

type SafeCounter struct {
    mu    sync.Mutex
    count int
}

func (c *SafeCounter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock() // unlocks when Increment returns, even on panic
    c.count++
}

func (c *SafeCounter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count
}

func main() {
    c := &SafeCounter{}
    var wg sync.WaitGroup

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            c.Increment()
        }()
    }
    wg.Wait()
    fmt.Println(c.Value()) // always 1000 — never a race condition
}

sync.RWMutex — Optimise Read-Heavy Workloads

When reads vastly outnumber writes, a plain Mutex creates unnecessary contention — only one reader can proceed at a time even though concurrent reads are perfectly safe. RWMutex allows any number of readers to hold the lock simultaneously; a writer gets exclusive access. The result is much higher throughput for read-heavy caches, configuration stores, and registries.

type Cache struct {
    mu    sync.RWMutex
    items map[string]string
}

func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock() // multiple goroutines can RLock simultaneously
    defer c.mu.RUnlock()
    v, ok := c.items[key]
    return v, ok
}

func (c *Cache) Set(key, val string) {
    c.mu.Lock() // exclusive — blocks all readers and other writers
    defer c.mu.Unlock()
    c.items[key] = val
}

sync/atomic — Lock-Free Operations

Atomic operations are the fastest way to update a single shared variable. They use CPU-level instructions (like LOCK XADD) that are indivisible — no mutex is needed, and there is no scheduler overhead. Use atomic for counters, flags, and state machines where a single variable is all you need to protect. For anything more complex (updating two related values, managing a data structure), reach for a mutex.

import "sync/atomic"

var requestCount int64

// Safe for concurrent use — no lock needed, faster than a mutex
func handleHTTP() {
    atomic.AddInt64(&requestCount, 1)
    // handle request...
}

func metrics() {
    count := atomic.LoadInt64(&requestCount) // atomic read — consistent view
    fmt.Printf("requests served: %d\n", count)
}

// Compare-and-swap for implementing lock-free state machines
// Only one goroutine can win the CAS — the others see false and back off
var state int32 // 0=idle, 1=running, 2=stopped

func startOnce() bool {
    return atomic.CompareAndSwapInt32(&state, 0, 1) // atomically: if state==0, set to 1
}

sync.Once — One-Time Initialization

sync.Once guarantees that a function runs exactly once, even when called concurrently from many goroutines. It’s the safe, idiomatic way to implement lazy initialization of expensive resources like database connection pools, caches, or parsed configuration. The first call runs the function; all subsequent calls (including concurrent ones) block until the first completes, then return immediately.

var (
    instance *Database
    once     sync.Once
)

func GetDB() *Database {
    once.Do(func() {
        // This runs exactly once, no matter how many goroutines call GetDB
        instance = &Database{
            pool: createConnectionPool(),
        }
    })
    return instance // always the same instance after the first call
}

Semaphore Pattern

A buffered channel makes an excellent semaphore for limiting concurrency. By sizing the buffer to the maximum number of simultaneous operations allowed, you guarantee that at most N goroutines are active at any moment — all others block until a slot is released. This is the standard Go approach for throttling parallel HTTP requests, database queries, or any I/O-bound work.

func parallelFetch(urls []string, maxConcurrent int) []string {
    sem := make(chan struct{}, maxConcurrent) // semaphore — limits to maxConcurrent goroutines
    results := make([]string, len(urls))
    var wg sync.WaitGroup

    for i, url := range urls {
        wg.Add(1)
        go func(idx int, u string) {
            defer wg.Done()
            sem <- struct{}{}        // acquire a slot — blocks if all slots are taken
            defer func() { <-sem }() // release the slot when done

            resp, err := http.Get(u)
            if err != nil {
                results[idx] = fmt.Sprintf("ERROR: %v", err)
                return
            }
            defer resp.Body.Close()
            results[idx] = fmt.Sprintf("%d %s", resp.StatusCode, u)
        }(i, url)
    }

    wg.Wait()
    return results
}

errgroup — WaitGroup with Error Propagation

sync.WaitGroup has no built-in way to collect errors from goroutines — you have to wire that up yourself. golang.org/x/sync/errgroup solves this: it combines WaitGroup semantics with automatic error capture. If any goroutine returns a non-nil error, Wait returns it. When paired with a context, the first error also cancels the context — stopping all other goroutines in the group cleanly.

import "golang.org/x/sync/errgroup"

func fetchAll(ctx context.Context, urls []string) error {
    // g is the group; ctx is derived from the parent and cancelled on first error
    g, ctx := errgroup.WithContext(ctx)

    for _, url := range urls {
        url := url // capture loop variable before the goroutine starts
        g.Go(func() error {
            req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
            if err != nil {
                return err
            }
            resp, err := http.DefaultClient.Do(req)
            if err != nil {
                return fmt.Errorf("fetching %s: %w", url, err)
            }
            defer resp.Body.Close()
            fmt.Printf("%d %s\n", resp.StatusCode, url)
            return nil
        })
    }

    return g.Wait() // blocks until all goroutines finish; returns the first error
}