Concurrency Patterns

Once you’re comfortable with goroutines and channels, a handful of composable patterns cover almost every real concurrency problem.

Worker pool

Bound the number of concurrent workers regardless of how many jobs arrive:

func workerPool(jobs <-chan int, results chan<- int, workerCount int) {
  var wg sync.WaitGroup
  for i := 0; i < workerCount; i++ {
    wg.Add(1)
    go func() {
      defer wg.Done()
      for j := range jobs {
        results <- j * j
      }
    }()
  }
  wg.Wait()
  close(results)
}

▶ Run

Pipelines

Chain stages together, each reading from one channel and writing to the next:

func generate(nums ...int) <-chan int {
  out := make(chan int)
  go func() {
    defer close(out)
    for _, n := range nums {
      out <- n
    }
  }()
  return out
}

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
}

// usage: for n := range square(generate(1, 2, 3)) { ... }

Each stage is independently testable and the whole pipeline runs concurrently — stage two starts squaring the first value while stage one is still generating the rest.

Fan-in

Merge several channels into one:

func fanIn(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
}

Avoiding leaks and races

• Always have a plan for how a channel gets closed — a goroutine blocked forever on a channel that’s never closed or read from is a leak.
• Run tests and real workloads with the race detector: go test -race ./... or go run -race main.go.
• Prefer passing a context.Context for cancellation over ad-hoc “done” channels once you have more than one exit condition (covered next chapter).

Use case

Bounded-concurrency fan-out over an API — e.g. fetching details for 500 IDs without hammering a rate-limited upstream — combines a worker pool with a semaphore-style buffered channel:

sem := make(chan struct{}, 10) // max 10 in flight
var wg sync.WaitGroup

for _, id := range ids {
  wg.Add(1)
  sem <- struct{}{}
  go func(id string) {
    defer wg.Done()
    defer func() { <-sem }()
    fetchDetails(id)
  }(id)
}
wg.Wait()