A worker pool finishes handing out jobs, so it seems safe to close the shared results channel right away and start reading from it. It isn't — the workers are still running when close(jobs) returns, since closing a channel only stops new sends into that channel; it says nothing about the results channel the workers are still writing to. Whichever worker is mid-send when results gets closed crashes the whole program.
func worker(id int, jobs <-chan int, results chan<- int) {
for j := range jobs {
time.Sleep(10 * time.Millisecond) // simulate work
results <- j * 2
}
}
jobs := make(chan int, 100)
results := make(chan int, 100)
for w := 1; w <= 3; w++ {
go worker(w, jobs, results)
}
for j := 1; j <= 9; j++ {
jobs <- j
}
close(jobs)
close(results) // closed immediately -- workers are still sending!
// panic: send on closed channelGo’s concurrency patterns are combinations of goroutines and channels rather than named framework constructs: a worker pool is a fixed number of goroutines pulling from a shared job channel; a pipeline chains stages together via channels, each stage a goroutine reading from one channel and writing to the next; fan-out spreads work across multiple goroutines, and fan-in merges multiple channels back into one.
Worker pool: a fixed number of goroutines sharing jobs
Rather than launching one goroutine per task (which could spawn thousands unpredictably), a worker pool launches a fixed number of goroutines that all read from the same jobs channel, naturally load-balancing work across them.
func worker(id int, jobs <-chan int, results chan<- int) {
for j := range jobs { // exits automatically when jobs is closed
results <- j * 2
}
}
jobs := make(chan int, 100)
results := make(chan int, 100)
for w := 1; w <= 3; w++ { // 3 workers, however many jobs
go worker(w, jobs, results)
}
for j := 1; j <= 9; j++ {
jobs <- j
}
close(jobs) // signals workers to stop once drainedPipeline: chaining stages with channels
Each stage is a function that takes an input channel and returns an output channel, run as its own goroutine — data flows through the pipeline one stage at a time, and each stage can start on later items before earlier ones finish downstream.
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
}
for result := range square(generate(1, 2, 3, 4)) {
fmt.Println(result) // 1, 4, 9, 16
}Fan-out, fan-in
Fan-out means multiple goroutines reading from the same channel to parallelize work (as the worker pool does above); fan-in is the reverse — merging several channels into one, typically with a helper that launches a goroutine per input channel, each forwarding to a shared output channel, closed via a sync.WaitGroup once all inputs are drained.
That sync.WaitGroup is exactly what the worker pool above was missing: each worker calls wg.Done() when it returns, and the goroutine that closes results calls wg.Wait() first, so the close can never race a worker that's still sending.
func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
defer wg.Done()
for j := range jobs {
time.Sleep(10 * time.Millisecond) // simulate work
results <- j * 2
}
}
jobs := make(chan int, 100)
results := make(chan int, 100)
var wg sync.WaitGroup
for w := 1; w <= 3; w++ {
wg.Add(1)
go worker(w, jobs, results, &wg)
}
for j := 1; j <= 9; j++ {
jobs <- j
}
close(jobs)
wg.Wait() // wait for every worker to finish sending
close(results) // now safe -- nothing is still writing
// sum: 90, every run