A worker pool runs a fixed number of goroutines pulling from a shared jobs channel, capping concurrency instead of spawning one goroutine per job.
Examples
Three workers share one jobs channel. Each pulls the next available job, so work distributes itself without a scheduler.
jobs := make(chan int, 5)
results := make(chan int, 5)
var wg sync.WaitGroup
for w := 1; w <= 3; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := range jobs {
results <- j * j
}
}()
}
for i := 1; i <= 5; i++ {
jobs <- i
}
close(jobs)
wg.Wait()
close(results)
sum := 0
for r := range results {
sum += r
}
fmt.Println(sum)55Job count and worker count are independent. Five workers process a thousand jobs the same way they process five, only the queue backs up longer, instead of spawning a thousand goroutines at once.
const numWorkers = 5
jobs := make(chan int, 1000)
var wg sync.WaitGroup
for w := 0; w < numWorkers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for range jobs {
}
}()
}
for i := 0; i < 1000; i++ {
jobs <- i
}
close(jobs)
wg.Wait()
fmt.Println("done")doneA result struct carries a value or an error, so a bad job doesn't stop the pool. Summing instead of printing per-result keeps the output order-independent.
type result struct {
value int
err error
}
jobs := make(chan int, 3)
results := make(chan result, 3)
var wg sync.WaitGroup
for w := 0; w < 2; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := range jobs {
if j == 0 {
results <- result{err: fmt.Errorf("invalid input: %d", j)}
continue
}
results <- result{value: j * 2}
}
}()
}
for _, j := range []int{0, 1, 2} {
jobs <- j
}
close(jobs)
wg.Wait()
close(results)
var sum, errCount int
for r := range results {
if r.err != nil {
errCount++
continue
}
sum += r.value
}
fmt.Println(sum, errCount)6 1ctx.Done() gives every worker and the job producer a shared stop signal, so a timeout or an unrelated failure shuts down the whole pool without a separate stop channel.
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
jobs := make(chan int)
var wg sync.WaitGroup
for w := 0; w < 3; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case _, ok := <-jobs:
if !ok {
return
}
case <-ctx.Done():
return
}
}
}()
}
go func() {
for i := 0; ; i++ {
select {
case jobs <- i:
case <-ctx.Done():
close(jobs)
return
}
}
}()
wg.Wait()
fmt.Println(ctx.Err())context deadline exceeded| Function | Description |
|---|---|
jobs := make(chan T, n) jobs := make(chan int, 100) | Buffered channel workers pull jobs from |
results := make(chan R, n) results := make(chan int, 100) | Channel workers send finished results to |
close(jobs) close(jobs) | Signals no more jobs; range over jobs ends once drained |
sync.WaitGroup wg.Add(1) / wg.Wait() | Tracks running workers so the caller can wait for them |
for j := range jobs for j := range jobs { ... } | Worker loop; exits automatically when jobs is closed and empty |
context.Context ctx.Done() | Shared cancellation signal for every worker and the producer |