Engineering

Go: Building a Worker Pool in a Controlled Way

Learn how to implement efficient worker pools in Go with proper concurrency control and error handling.

January 20, 2024•6 min read
Go: Building a Worker Pool in a Controlled Way

Understanding Worker Pools in Go

Worker pools are a fundamental concurrency pattern in Go that allow you to control the number of goroutines executing tasks concurrently. This pattern is essential when you need to process a large number of jobs while limiting resource consumption and preventing system overload.

In this guide we'll explore how to implement robust worker pools with controlled concurrency, proper error handling, and graceful shutdown capabilities.

Basic Worker Pool Implementation

Start with a simple worker pool that processes jobs concurrently using a fixed number of workers. The basic pattern consists of a job queue, worker goroutines, and result collection.

go
package main

    import (
      "fmt"
      "time"
    )

    type Job struct {
      ID       int
      Payload  string
      Duration time.Duration
    }

    type Result struct {
      JobID    int
      Success  bool
      Duration time.Duration
    }

    type Worker struct {
      ID       int
      JobQueue chan Job
      Results  chan Result
      Quit     chan bool
    }

    func NewWorker(id int, jq chan Job, res chan Result) *Worker {
      return &Worker{ID: id, JobQueue: jq, Results: res, Quit: make(chan bool)}
    }

    func (w *Worker) Start() {
      go func() {
        for {
          select {
          case job := <-w.JobQueue:
            start := time.Now()
            // process job
            time.Sleep(job.Duration)
            w.Results <- Result{JobID: job.ID, Success: true, Duration: time.Since(start)}
          case <-w.Quit:
            return
          }
        }
      }()
    }

Each worker listens for jobs on a channel, processes them, and sends results back through a result channel. Workers can be signalled to stop for graceful shutdowns.

Worker Pool Manager

Create a manager that coordinates multiple workers and provides a clean interface for submitting jobs and collecting results.

go
type WorkerPool struct {
      JobQueue chan Job
      Results  chan Result
      Workers  []*Worker
    }

    func NewWorkerPool(numWorkers int) *WorkerPool {
      jq := make(chan Job, numWorkers*2)
      res := make(chan Result, numWorkers*2)

      wp := &WorkerPool{JobQueue: jq, Results: res}
      for i := 0; i < numWorkers; i++ {
        w := NewWorker(i+1, jq, res)
        wp.Workers = append(wp.Workers, w)
        w.Start()
      }

      return wp
    }

    func (wp *WorkerPool) Submit(job Job) {
      wp.JobQueue <- job
    }

    func (wp *WorkerPool) Shutdown() {
      for _, w := range wp.Workers {
        w.Quit <- true
      }
      close(wp.JobQueue)
      close(wp.Results)
    }

The WorkerPool handles job distribution, result collection, and graceful shutdown. Use buffered channels to avoid blocking under load.

Advanced Features

Enhance the pool with rate limiting, circuit breaking, metrics, and timeouts for production readiness.

go
// Example: adding a rate limiter and context-aware jobs
    import "context"

    type ContextJob struct {
      Job
      Ctx context.Context
    }

    // Submit with context and timeout
    func (wp *WorkerPool) SubmitWithContext(ctx context.Context, job Job) error {
      select {
      case <-ctx.Done():
        return ctx.Err()
      case wp.JobQueue <- job:
        return nil
      }
    }

    // Example rate limiter using a ticker
    func RateLimitedSubmit(wp *WorkerPool, job Job, ratePerSecond int) {
      ticker := time.NewTicker(time.Second / time.Duration(ratePerSecond))
      defer ticker.Stop()

      <-ticker.C
      wp.Submit(job)
    }

These features improve resilience and observability in real-world systems.

Best Practices

✅ Do's

  • Use buffered channels for job queues
  • Implement graceful shutdown mechanisms
  • Add logging and metrics
  • Use context for cancellation
  • Monitor resource usage

❌ Don'ts

  • Create unbounded goroutines
  • Ignore errors in workers
  • Use blocking operations without timeouts
  • Forget to close channels
;