Worker Pools and Fan-In
A fixed set of goroutines reading one jobs channel bounds your concurrency, and one results channel collects their output.
for i := 0; i < workers; i++ { go func() { for j := range jobs { results <- do(j) } }() }Starting a goroutine per job works until jobs arrive faster than they finish. Ten thousand goroutines each opening a database connection will exhaust something. A worker pool fixes the number of goroutines instead.
The shape is two channels. One carries jobs, one carries results. You start a fixed number of goroutines, and each loops over the jobs channel with range. Whichever worker is free takes the next job. The worker count, not the job count, is the limit on how much happens at once. That limit is the whole point: it is what stops a burst of work from overwhelming a database, an API quota, or memory.
Knowing when it is over
The producer closes the jobs channel after the last job. Every worker's range loop then ends and the worker returns.
The results channel is harder, because no single worker knows the others are done. Use a sync.WaitGroup: one goroutine calls Wait and closes the results channel after that, so the collecting range loop can end too.
wg.Go(f), added in Go 1.25, starts f in a goroutine and counts it. The longhand is wg.Add(1) followed by go func() { defer wg.Done(); f() }(). They are the same thing.
Fan-in, and what trips people up
Many workers sending on one channel is fan-in: several producers, one receiver, and the receiver does not care which worker sent what.
Results arrive in whatever order the workers finish. If order matters, send an index alongside each result, or sort afterwards.
Example
package main
import (
"fmt"
"slices"
"sync"
)
func main() {
jobs := make(chan int)
results := make(chan int)
// Three workers means at most three jobs run at once, however many
// jobs arrive. The worker count is the bound on concurrency.
var wg sync.WaitGroup
for w := 0; w < 3; w++ {
wg.Go(func() {
for n := range jobs { // ends when jobs is closed
results <- n * n
}
})
}
go func() {
for n := 1; n <= 5; n++ {
jobs <- n
}
close(jobs) // every worker's range loop now finishes
}()
go func() {
wg.Wait() // all workers have stopped sending
close(results) // so the collecting loop below can end
}()
var squares []int
for r := range results { // fan-in: many senders, one receiver
squares = append(squares, r)
}
slices.Sort(squares) // workers finish in any order
fmt.Println("squares:", squares)
}
// Output:
// squares: [1 4 9 16 25]When to use it
- Resize 5,000 uploaded images with eight workers instead of 5,000 goroutines.
- Call a rate-limited API from a fixed number of goroutines so you stay under the quota.
- Bound concurrent database work to the size of the connection pool.
- Merge results from several shards into one response-building loop.
More examples
A pool with a known result count and no WaitGroup
The simplest correct pool. When every job produces exactly one result and you know the job count in advance, you can count the results instead of coordinating the workers, which removes the <code>WaitGroup</code> and the extra goroutine that closes <code>results</code>. Both channels are buffered to the job count so neither the producer nor a worker ever blocks. Use this version whenever the work is a fixed batch: resize these 200 images, fetch these 50 URLs. Switch to the <code>WaitGroup</code> version when jobs keep arriving, or when a job can produce no result at all.
package main
import (
"fmt"
"slices"
)
func main() {
const jobCount = 6
const workers = 2
jobs := make(chan int, jobCount)
results := make(chan int, jobCount)
// Two workers, six jobs: at most two squarings happen at a time.
for w := 0; w < workers; w++ {
go func() {
for n := range jobs { // ends when jobs is closed and empty
results <- n * n
}
}()
}
for n := 1; n <= jobCount; n++ {
jobs <- n
}
close(jobs)
// Exactly jobCount results are coming, so counting them is enough.
// No WaitGroup is needed to know when to stop collecting.
got := make([]int, 0, jobCount)
for i := 0; i < jobCount; i++ {
got = append(got, <-results)
}
slices.Sort(got)
fmt.Println("squares:", got)
}
// Output:
// squares: [1 4 9 16 25 36]Fan-in: merging several channels into one
Fan-in as a reusable function. <code>merge</code> takes any number of receive-only channels and returns one, so a caller reads a single stream with one <code>range</code> loop and does not care how many sources there were. One goroutine per input copies values across; the closing goroutine waits for all of them. This beats a <code>select</code> with a fixed number of cases, because the number of inputs does not have to be known when you write the code. Reach for it when several producers feed one consumer: shards of a query, replicas of a feed, partitions of a topic.
package main
import (
"fmt"
"sync"
)
// merge copies every value from every input onto one output channel, and
// closes that output once all the inputs are done. This is fan-in.
func merge(inputs ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
for _, in := range inputs {
wg.Go(func() {
for v := range in {
out <- v
}
})
}
go func() {
wg.Wait()
close(out)
}()
return out
}
// source hands back a closed channel already holding values.
func source(values ...int) <-chan int {
ch := make(chan int, len(values))
for _, v := range values {
ch <- v
}
close(ch)
return ch
}
func main() {
total := 0
for v := range merge(source(1, 2), source(3, 4), source(5)) {
total += v
}
fmt.Println("total:", total)
}
// Output:
// total: 15wg.Go and the longhand it replaces
Both forms in one program, doing the same job. <code>wg.Go</code> arrived in Go 1.25, so every pool written before that uses the three-line version, and you will keep meeting it. The longhand also shows why the shorthand is worth having: <code>Add(1)</code> must run in the parent, before the goroutine starts, and <code>Done</code> must run on every exit path including a panic, which is what the <code>defer</code> guarantees. Calling <code>Add</code> inside the new goroutine is the classic bug, because <code>Wait</code> can then return before that goroutine has even been counted.
package main
import (
"fmt"
"sync"
)
func main() {
out := make(chan string, 2)
var wg sync.WaitGroup
// Shorthand, added in Go 1.25.
wg.Go(func() {
out <- "short"
})
// Longhand: what wg.Go does, written out. Add runs before the goroutine
// starts, and defer makes Done run on every way out of the function.
wg.Add(1)
go func() {
defer wg.Done()
out <- "long"
}()
wg.Wait()
close(out)
count := 0
for range out {
count++
}
fmt.Println("values collected:", count)
}
// Output:
// values collected: 2
Discussion