Go Worker Pools — Managing Concurrent Work with Worker Goroutines
In this tutorial, you will learn about Go Worker Pools. We cover key concepts, practical examples, and best practices to help you master this topic.
Go worker pools manage concurrent work using a fixed number of goroutines consuming jobs from a channel with results collected via sync.WaitGroup.
What You'll Learn
- Fixed-size worker pool pattern
- Dynamic worker pool with context
- Error handling in worker pools
- Rate limiting with ticker
Why It Matters
Worker pools prevent resource exhaustion. Docker uses pools for image layer processing. Kubernetes uses pools for API request handling. DodaZIP uses pools for file compression jobs.
Real-World Use
Image processing pipelines, batch file conversion, API request batching, database migration workers, email sending queues.
flowchart LR
A["Worker Pools"] --> B["Jobs Channel"]
A --> C["Workers"]
A --> D["Results"]
B --> C
C --> D
A:::current --> B
style A fill:#2563eb,stroke:#2563eb,color:#fff
style B fill:#dbeafe,stroke:#2563eb,color:#1e40af
style C fill:#dbeafe,stroke:#2563eb,color:#1e40af
style D fill:#f1f5f9,stroke:#94a3b8,color:#64748b
Basic Worker Pool
func worker(id int, jobs <-chan int, results chan<- int) {
for job := range jobs {
fmt.Printf("Worker %d processing job %d\n", id, job)
time.Sleep(time.Second)
results <- job * 2
}
}
func main() {
const numJobs = 10
const numWorkers = 3
jobs := make(chan int, numJobs)
results := make(chan int, numJobs)
for w := 1; w <= numWorkers; w++ {
go worker(w, jobs, results)
}
for j := 1; j <= numJobs; j++ {
jobs <- j
}
close(jobs)
for r := 1; r <= numJobs; r++ {
<-results
}
}
Worker Pool with WaitGroup
func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
fmt.Printf("Worker %d processing %d\n", id, job)
time.Sleep(time.Second)
}
}
func main() {
jobs := make(chan int, 10)
var wg sync.WaitGroup
for w := 1; w <= 3; w++ {
wg.Add(1)
go worker(w, jobs, &wg)
}
for j := 1; j <= 10; j++ {
jobs <- j
}
close(jobs)
wg.Wait()
}
Pool with Error Handling
type Result struct {
Value int
Err error
}
func worker(id int, jobs <-chan int, results chan<- Result) {
for job := range jobs {
if job%3 == 0 {
results <- Result{Err: fmt.Errorf("job %d failed", job)}
continue
}
results <- Result{Value: job * 2}
}
}
func main() {
jobs := make(chan int, 10)
results := make(chan Result, 10)
for w := 1; w <= 3; w++ {
go worker(w, jobs, results)
}
for j := 1; j <= 10; j++ {
jobs <- j
}
close(jobs)
for r := 1; r <= 10; r++ {
res := <-results
if res.Err != nil {
fmt.Println("Error:", res.Err)
} else {
fmt.Println("Result:", res.Value)
}
}
}
Rate-Limited Pool
func main() {
jobs := make(chan int, 10)
results := make(chan int, 10)
for w := 1; w <= 3; w++ {
go func(id int) {
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
for job := range jobs {
<-ticker.C
fmt.Printf("Worker %d: processing %d\n", id, job)
results <- job * 2
}
}(w)
}
for j := 1; j <= 10; j++ {
jobs <- j
}
close(jobs)
for r := 1; r <= 10; r++ {
<-results
}
}
Context-Controlled Pool
func worker(id int, jobs <-chan int, results chan<- int, ctx context.Context) {
for {
select {
case <-ctx.Done():
fmt.Printf("Worker %d shutting down\n", id)
return
case job, ok := <-jobs:
if !ok { return }
fmt.Printf("Worker %d processing %d\n", id, job)
time.Sleep(time.Second)
results <- job * 2
}
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
jobs := make(chan int, 10)
results := make(chan int, 10)
for w := 1; w <= 3; w++ {
go worker(w, jobs, results, ctx)
}
for j := 1; j <= 10; j++ {
jobs <- j
}
close(jobs)
for r := 1; r <= 10; r++ {
select {
case res := <-results:
fmt.Println("Result:", res)
case <-ctx.Done():
fmt.Println("Timeout reached")
return
}
}
}
Common Mistakes
1. Not Closing Jobs Channel
Workers range over jobs. If you don't close, they wait forever.
2. Wrong Buffer Size
Jobs channel should buffer to avoid blocking senders. Results channel matching job count prevents sender Goroutine leaks.
3. Too Many Workers
More workers than CPU cores can hurt performance for CPU-bound work. Measure and tune.
4. Not Handling Worker Panics
Workers should recover from panics. Use defer/recover in worker functions.
5. Ignoring Context Cancellation
Long-running workers should check ctx.Done() for clean shutdown.
Practice Questions
1. What determines optimal worker count? For I/O-bound: number of concurrent connections. For CPU-bound: number of CPU cores. Always profile.
2. When should workers send results? After each job for real-time results. After batch completion for bulk processing.
3. How do you stop workers gracefully? Close the jobs channel and check ctx.Done(). Workers exit when jobs channel is exhausted or context is cancelled.
4. What happens if results channel fills up? Workers block on sending results. Ensure results channel has sufficient buffer or Process results concurrently.
Challenge: Implement a worker pool for downloading URLs concurrently with 5 workers.
Solution
type DownloadResult struct {
URL string
Size int64
Err error
}
func downloader(jobs <-chan string, results chan<- DownloadResult) {
for url := range jobs {
resp, err := http.Get(url)
if err != nil {
results <- DownloadResult{URL: url, Err: err}
continue
}
results <- DownloadResult{URL: url, Size: resp.ContentLength}
resp.Body.Close()
}
}
FAQ
{{< faq question="What is the difference between worker pool and fan-out?" >}} Worker pool has fixed workers consuming from a shared channel. Fan-out distributes work across goroutines on creation. {{< /faq >}}
{{< faq question="Should workers send errors on results channel?" >}} Yes. Include errors in results rather than using a separate error channel. Use a result struct with error field. {{< /faq >}}
{{< faq question="How do I implement a dynamic pool?" >> Use a manager goroutine that spawns workers as needed. Monitor queue depth and adjust pool size. {{< /faq >}}
{{< faq question="Can workers share state?" >}} Workers should be stateless. If they need shared state, use a mutex or pass data through channels. {{< /faq >}}
{{< faq question="How do I test worker pools?" >}} Test worker logic in isolation. Test pool Orchestration with controlled inputs. Use context timeouts in tests for safety. {{< /faq >}}
Try It Yourself
package main
import (
"fmt"
"sync"
)
func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
fmt.Printf("Worker %d: %d^2 = %d\n", id, job, job*job)
}
}
func main() {
jobs := make(chan int, 10)
var wg sync.WaitGroup
for w := 1; w <= 3; w++ {
wg.Add(1)
go worker(w, jobs, &wg)
}
for j := 1; j <= 6; j++ {
jobs <- j
}
close(jobs)
wg.Wait()
}
Expected output:
Worker 1: 1^2 = 1
Worker 2: 2^2 = 4
Worker 3: 3^2 = 9
Worker 1: 4^2 = 16
Worker 2: 5^2 = 25
Worker 3: 6^2 = 36
What's Next
Now that you understand worker pools, explore advanced concurrency patterns in Go.
| Topic | Description | Link |
|---|---|---|
| Go Concurrency Patterns | Advanced patterns | {{< ref "24-concurrency-patterns" >}} |
| Go Channels | Channel communication | {{< ref "18-channels" >}} |
| Go Context | Cancellation and deadlines | {{< ref "22-context" >}} |