Goroutines: the Worker Pool pattern

Hey everyone! This is the first post in a series I’m excited to share about concurrency models in Go. To kick things off I picked a model known as the Worker Pool. But what exactly is a worker pool?

Basically, it’s a pattern where a fixed set of goroutines — also called workers — wait to receive tasks to execute. This model brings several advantages and disadvantages, but before we dive into them, let’s visualize how it works.

How it works

In this basic example, we have a web application in Go that implements a worker pool. When someone calls the application, our software doesn’t execute the task immediately. Instead, it stores the task in a queue and returns a response to the user. Meanwhile, in another goroutine, the workers — acting as consumers — listen for new tasks on the queue and process them as capacity becomes available.

Benefits of a worker pool

So what’s the benefit of this model? One of the biggest advantages is preventing a goroutine leak. Imagine we’re operating with a limited amount of RAM/CPU and a low RPS (requests per second), say 30 RPS. If each task takes about 20 seconds to process, we’d only ever have around 600 goroutines open at once, which is manageable and wouldn’t heavily strain our infrastructure.

Now suppose there’s a usage spike and our RPS jumps to 60. We’d be looking at a minimum of 1200 goroutines processing at the same time. That could become unsustainable at some point. The worker pool model addresses this elegantly, because it lets us define a fixed processing capacity. So instead of dealing with a goroutine leak during peak times, the application simply slows down. That backpressure trips our alerts, prompting us to increase container capacity and the number of workers.

Drawbacks of a worker pool

Of course, nothing in software is perfect. If you’ve read this far you may have already thought of a few drawbacks — such as the risk of sizing the pool incorrectly, leaving us consistently below the task count. That would make processing slower than just spawning a goroutine per task.

Another important point is how we build the task queue. In critical processing scenarios we might need a way to persist tasks so they aren’t lost if the application crashes. And there’s the usual concurrency challenge of monitoring the execution state of those tasks. Well-structured code with proper logs goes a long way here, but it’s still worth discussing.

Time to code

Enough talking — let’s build a small example to play with. We’ll start with the structs that make the worker pool work. First, the Worker:

type Worker struct {
    ID        int
    TaskQueue chan Task
}

Our worker has two fields: an ID for identification and a channel it listens on for new tasks to process.

type WorkerPool struct {
    Workers   []*Worker
    TaskQueue chan Task
}

type Task struct {
    ID string
}

Next we create the WorkerPool, which manages the workers and uses the task channel to notify them, along with a simple Task struct.

After that, a function that builds the pool:

func NewWorkerPool(numWorkers int, wg *sync.WaitGroup) *WorkerPool {
    taskQueue := make(chan Task, 1000) // a buffered channel

    pool := &WorkerPool{
        Workers:   make([]*Worker, numWorkers),
        TaskQueue: taskQueue,
    }

    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        pool.Workers[i] = NewWorker(i+1, wg, taskQueue)
    }

    return pool
}

This function takes the number of workers we need and creates a buffered taskQueue. What does that 1000 mean? It’s the maximum number of tasks the channel can hold. If we exceed it, the channel won’t error — senders simply block until space frees up, letting the workers keep processing.

Now let’s write NewWorker:

func NewWorker(id int, wg *sync.WaitGroup, taskQueue chan Task) *Worker {
    worker := &Worker{
        ID:        id,
        TaskQueue: taskQueue,
    }

    go worker.start(wg)

    return worker
}

func (w *Worker) start(wg *sync.WaitGroup) {
    defer wg.Done()

    for task := range w.TaskQueue {
        fmt.Printf("Worker %d processing task ID %s\n", w.ID, task.ID)
        time.Sleep(10 * time.Second)
        fmt.Printf("Worker %d finished task ID %s\n", w.ID, task.ID)
    }
}

NewWorker creates a worker from the given attributes — the ID and the task channel — and calls .start() to put it to work. The start method ranges over the channel, processing each task as it arrives, and exits cleanly once the channel is closed.

To make this functional, we add a handler that pushes new tasks onto the channel for the workers to process.

var pool *WorkerPool

func main() {
    numWorkers := 4
    wg := &sync.WaitGroup{}
    pool = NewWorkerPool(numWorkers, wg)

    stop := make(chan os.Signal, 1)
    signal.Notify(stop, os.Interrupt, syscall.SIGTERM)

    server := &http.Server{Addr: ":8081"}
    http.HandleFunc("GET /task/{id}", taskHandler)

    go func() {
        fmt.Println("Server running on :8081")
        if err := server.ListenAndServe(); err != http.ErrServerClosed {
            fmt.Printf("server error: %v\n", err)
        }
    }()

    <-stop
    server.Shutdown(context.Background())
    close(pool.TaskQueue)
    wg.Wait()
    fmt.Println("Shutdown complete")
}

func taskHandler(w http.ResponseWriter, r *http.Request) {
    taskID := r.PathValue("id")

    pool.TaskQueue <- Task{ID: taskID}
    fmt.Fprintf(w, "Task ID %s added\n", taskID)
}

Note the stop channel, which listens for shutdown signals. When one arrives we stop the server, close the task channel so no new tasks are accepted, and wait for the workers to finish whatever they’re still processing. Only then does the application exit — that’s our graceful shutdown.

Run the application and make a GET request to the endpoint:

curl http://localhost:8081/task/1

And there you have it — our worker pool is up and processing tasks. As you can see, there’s no fixed order for which worker picks up which task; in fact, Worker 2 grabs the first one.

I hope you enjoyed the post! Let’s keep exploring Go and its possibilities.

• GitHub gist with the complete code