Go Worker Pool: Concurrent Task Processing Patterns for High-Load Systems

Introduction: Why You Need a Worker Pool in Go Services

In 2026, Go remains one of the dominant languages for building high-load backend systems. Goroutines are cheap — you can spawn millions of them — but that doesn't mean uncontrolled goroutine spawning is the right strategy. In real-world services, unbounded concurrency leads to file descriptor exhaustion, garbage collector pressure, overload of downstream dependencies (databases, third-party APIs), and unpredictable latency.

A worker pool addresses several key challenges:

Concurrency limiting — no more than N tasks execute simultaneously.
Goroutine reuse — instead of creating a goroutine per request, a fixed pool is used.
Backpressure — when the queue is full, the producer blocks or receives an error rather than generating load uncontrollably.
Error isolation — a panic in one worker won't crash the entire program when a proper recover is in place.

In this article, we'll walk through everything step by step: from a primitive implementation to a production-ready solution with dynamic scaling, retry logic, graceful shutdown, PostgreSQL integration as a reliable queue, and full observability.

Basic Worker Pool Implementation with Goroutines and Channels

A classic worker pool in Go is built on three primitives: a jobs channel (jobs chan), a results channel (results chan), and sync.WaitGroup to wait for all workers to finish.

package workerpool

import (
    "context"
    "fmt"
    "sync"
)

// Job represents a unit of work to be executed.
type Job struct {
    ID      int
    Payload any
}

// Result holds the outcome of a job execution.
type Result struct {
    JobID int
    Value any
    Err   error
}

// ProcessFunc is the job processing function passed when creating the pool.
type ProcessFunc func(ctx context.Context, job Job) Result

// Pool is a basic worker pool.
type Pool struct {
    jobs    chan Job
    results chan Result
    wg      sync.WaitGroup
    process ProcessFunc
    size    int
}

// NewPool creates a pool with size workers and a queue buffer of queueSize.
func NewPool(size, queueSize int, fn ProcessFunc) *Pool {
    return &Pool{
        jobs:    make(chan Job, queueSize),
        results: make(chan Result, queueSize),
        process: fn,
        size:    size,
    }
}

// Start launches the workers and begins processing jobs.
func (p *Pool) Start(ctx context.Context) {
    for i := 0; i < p.size; i++ {
        p.wg.Add(1)
        go p.worker(ctx)
    }
}

// worker is a goroutine that reads jobs from the jobs channel.
func (p *Pool) worker(ctx context.Context) {
    defer p.wg.Done()
    for {
        select {
        case job, ok := <-p.jobs:
            if !ok {
                // Channel closed — worker exits.
                return
            }
            result := p.process(ctx, job)
            // Send the result; block if the channel is full.
            select {
            case p.results <- result:
            case <-ctx.Done():
                return
            }
        case <-ctx.Done():
            return
        }
    }
}

// Submit sends a job to the queue.
// Returns an error if the context is cancelled or the channel is full.
func (p *Pool) Submit(ctx context.Context, job Job) error {
    select {
    case p.jobs <- job:
        return nil
    case <-ctx.Done():
        return fmt.Errorf("worker pool: submit cancelled: %w", ctx.Err())
    }
}

// Results returns the results channel for the consumer to read from.
func (p *Pool) Results() <-chan Result {
    return p.results
}

// Stop closes the jobs channel and waits for all workers to finish.
func (p *Pool) Stop() {
    close(p.jobs)
    p.wg.Wait()
    close(p.results)
}

Key architectural decisions in this implementation:

The jobs channel is buffered — the producer won't block immediately during brief load spikes.
Each worker listens to two channels via select: jobs and context cancellation — this prevents hangs during shutdown.
Closing jobs signals workers to stop: Go guarantees that all previously sent values will be read after the channel is closed.

Dynamic Pool Scaling: Adapting to Load

A static pool is optimal for predictable load. In real systems, load varies: traffic is minimal at night and ten times higher during peak hours. A dynamic pool allows scaling the number of workers within a [minWorkers, maxWorkers] range based on a metric — the length of the job queue.

package workerpool

import (
    "context"
    "sync"
    "sync/atomic"
    "time"
)

// DynamicPool scales the number of workers based on load.
type DynamicPool struct {
    jobs       chan Job
    results    chan Result
    process    ProcessFunc
    mu         sync.Mutex
    wg         sync.WaitGroup
    ctx        context.Context
    cancel     context.CancelFunc
    active     atomic.Int64 // current number of workers
    minWorkers int
    maxWorkers int
}

func NewDynamicPool(min, max, queueSize int, fn ProcessFunc) *DynamicPool {
    ctx, cancel := context.WithCancel(context.Background())
    p := &DynamicPool{
        jobs:       make(chan Job, queueSize),
        results:    make(chan Result, queueSize),
        process:    fn,
        ctx:        ctx,
        cancel:     cancel,
        minWorkers: min,
        maxWorkers: max,
    }
    // Start the minimum number of workers immediately.
    for i := 0; i < min; i++ {
        p.startWorker()
    }
    // Start the autoscaler goroutine.
    go p.autoscale()
    return p
}

func (p *DynamicPool) startWorker() {
    p.active.Add(1)
    p.wg.Add(1)
    go func() {
        defer func() {
            p.active.Add(-1)
            p.wg.Done()
        }()
        for {
            select {
            case job, ok := <-p.jobs:
                if !ok {
                    return
                }
                result := p.process(p.ctx, job)
                select {
                case p.results <- result:
                case <-p.ctx.Done():
                    return
                }
            case <-p.ctx.Done():
                return
            }
        }
    }()
}

// autoscale checks the load every 500ms and adds workers when necessary.
func (p *DynamicPool) autoscale() {
    ticker := time.NewTicker(500 * time.Millisecond)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            queueLen := len(p.jobs)
            current := int(p.active.Load())
            // If the queue is more than 50% full and we can grow — add a worker.
            if queueLen > cap(p.jobs)/2 && current < p.maxWorkers {
                p.startWorker()
            }
            // If the queue is empty and we have more than the minimum — stop one.
            // Use a per-worker drain channel for stopping (simplified here).
        case <-p.ctx.Done():
            return
        }
    }
}

func (p *DynamicPool) Stop() {
    p.cancel()
    close(p.jobs)
    p.wg.Wait()
    close(p.results)
}

To reduce the number of workers in production implementations, use a separate quit channel per worker or the golang.org/x/sync/semaphore library for precise concurrency control.

Error Handling and Retry Logic Without Goroutine Leaks

In high-load systems, some jobs will inevitably fail: network errors, temporary dependency unavailability, context deadlines. A naive retry loop inside a worker blocks it and reduces pool throughput. The correct approach is to resubmit the job to the queue with exponential backoff and a limited number of attempts.

package workerpool

import (
    "context"
    "errors"
    "log/slog"
    "math"
    "time"
)

// RetryableJob extends Job with retry metadata.
type RetryableJob struct {
    Job
    Attempt    int
    MaxRetries int
    NextRunAt  time.Time
}

// withRetry wraps a ProcessFunc to add retry logic.
func withRetry(pool *Pool, fn ProcessFunc) ProcessFunc {
    return func(ctx context.Context, job Job) Result {
        rj, ok := job.Payload.(RetryableJob)
        if !ok {
            return fn(ctx, job)
        }

        result := fn(ctx, job)
        if result.Err == nil {
            return result
        }

        // Don't retry on context cancellation.
        if errors.Is(result.Err, context.Canceled) || errors.Is(result.Err, context.DeadlineExceeded) {
            return result
        }

        if rj.Attempt >= rj.MaxRetries {
            slog.Error("job failed permanently",
                "job_id", job.ID,
                "attempts", rj.Attempt,
                "err", result.Err,
            )
            return result
        }

        // Exponential backoff: 2^attempt seconds, capped at 60 seconds.
        delay := time.Duration(math.Min(math.Pow(2, float64(rj.Attempt)), 60)) * time.Second
        rj.Attempt++
        rj.NextRunAt = time.Now().Add(delay)

        // Schedule resubmission in a separate goroutine.
        // The goroutine exits after the delay or on context cancellation — no leak.
        go func() {
            select {
            case <-time.After(delay):
                retryJob := Job{ID: job.ID, Payload: rj}
                if err := pool.Submit(ctx, retryJob); err != nil {
                    slog.Warn("failed to resubmit job", "job_id", job.ID, "err", err)
                }
            case <-ctx.Done():
                slog.Info("retry cancelled due to context", "job_id", job.ID)
            }
        }()

        return Result{JobID: job.ID, Err: nil} // don't treat a transient error as final
    }
}

A critical point: the deferred retry goroutine always terminates — either after the timer fires or when the context is cancelled. This eliminates goroutine leaks — one of the most dangerous issues in long-running Go services.

Graceful Shutdown: Completing All In-Flight Tasks Cleanly

A production service must shut down cleanly upon receiving SIGTERM or SIGINT: finish already-accepted jobs, stop accepting new ones, and release resources.

package main

import (
    "context"
    "log/slog"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/yourorg/workerpool"
)

func main() {
    // Application context: cancelled when an OS signal is received.
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
    defer stop()

    pool := workerpool.NewPool(10, 100, processImage)
    pool.Start(ctx)

    // Producer goroutine: stops when the context is cancelled.
    go func() {
        for i := 0; ; i++ {
            select {
            case <-ctx.Done():
                slog.Info("producer stopped")
                return
            default:
            }
            job := workerpool.Job{ID: i, Payload: fmt.Sprintf("image_%d.jpg", i)}
            if err := pool.Submit(ctx, job); err != nil {
                slog.Warn("submit failed", "err", err)
            }
        }
    }()

    // Read results in a separate goroutine.
    go func() {
        for result := range pool.Results() {
            if result.Err != nil {
                slog.Error("job error", "job_id", result.JobID, "err", result.Err)
            }
        }
    }()

    // Wait for the shutdown signal.
    <-ctx.Done()
    slog.Info("shutdown signal received, draining pool...")

    // Give workers up to 30 seconds to finish processing.
    shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    done := make(chan struct{})
    go func() {
        pool.Stop() // closes the jobs channel and waits on WaitGroup
        close(done)
    }()

    select {
    case <-done:
        slog.Info("graceful shutdown complete")
    case <-shutdownCtx.Done():
        slog.Error("shutdown timeout exceeded, forcing exit")
        os.Exit(1)
    }
}

The signal.NotifyContext pattern was introduced in Go 1.16 and is the idiomatic way to tie an application's lifecycle to OS signals. The shutdown timeout (30 seconds here) prevents indefinite waiting if any jobs are stuck.

PostgreSQL Integration: A Reliable Job Queue

For jobs that cannot be lost on service restart (sending emails, processing payments), an in-memory channel queue is insufficient — data in memory is not persistent. PostgreSQL with the FOR UPDATE SKIP LOCKED pattern solves this problem, turning a table into a reliable distributed queue.

-- Jobs table schema
CREATE TABLE jobs (
    id          BIGSERIAL PRIMARY KEY,
    type        TEXT NOT NULL,
    payload     JSONB NOT NULL,
    status      TEXT NOT NULL DEFAULT 'pending', -- pending | running | done | failed
    attempts    INT NOT NULL DEFAULT 0,
    max_retries INT NOT NULL DEFAULT 3,
    run_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_jobs_status_run_at ON jobs (status, run_at)
    WHERE status IN ('pending', 'failed');

package pgqueue

import (
    "context"
    "encoding/json"
    "time"

    "github.com/jackc/pgx/v5"
    "github.com/jackc/pgx/v5/pgxpool"
)

type PGJob struct {
    ID         int64
    Type       string
    Payload    json.RawMessage
    Attempts   int
    MaxRetries int
}

type Queue struct {
    db *pgxpool.Pool
}

func NewQueue(db *pgxpool.Pool) *Queue {
    return &Queue{db: db}
}

// Dequeue atomically claims one job from the queue.
// FOR UPDATE SKIP LOCKED ensures concurrent workers never pick up the same job twice.
func (q *Queue) Dequeue(ctx context.Context) (*PGJob, error) {
    tx, err := q.db.Begin(ctx)
    if err != nil {
        return nil, err
    }
    defer tx.Rollback(ctx)

    var job PGJob
    err = tx.QueryRow(ctx, `
        UPDATE jobs
        SET status = 'running',
            attempts = attempts + 1,
            updated_at = NOW()
        WHERE id = (
            SELECT id FROM jobs
            WHERE status IN ('pending', 'failed')
              AND run_at <= NOW()
              AND attempts < max_retries
            ORDER BY run_at ASC
            FOR UPDATE SKIP LOCKED
            LIMIT 1
        )
        RETURNING id, type, payload, attempts, max_retries
    `).Scan(&job.ID, &job.Type, &job.Payload, &job.Attempts, &job.MaxRetries)

    if err != nil {
        if err == pgx.ErrNoRows {
            return nil, nil // queue is empty
        }
        return nil, err
    }

    return &job, tx.Commit(ctx)
}

// Complete marks a job as successfully finished.
func (q *Queue) Complete(ctx context.Context, jobID int64) error {
    _, err := q.db.Exec(ctx,
        `UPDATE jobs SET status = 'done', updated_at = NOW() WHERE id = $1`,
        jobID,
    )
    return err
}

// Fail marks a job as failed and schedules a retry with exponential backoff.
func (q *Queue) Fail(ctx context.Context, jobID int64, attempts int) error {
    delay := time.Duration(1<<attempts) * time.Second // 2^attempts seconds
    if delay > 10*time.Minute {
        delay = 10 * time.Minute
    }
    _, err := q.db.Exec(ctx, `
        UPDATE jobs
        SET status = 'failed',
            run_at = NOW() + $1::interval,
            updated_at = NOW()
        WHERE id = $2
    `, delay.String(), jobID)
    return err
}

The key property of FOR UPDATE SKIP LOCKED: rows locked by other transactions are skipped rather than causing a blocking wait. This makes the pattern highly scalable — dozens of workers can concurrently poll the same table without deadlocking each other.

For the worker polling loop, use adaptive pauses: if the queue is empty, gradually increase the polling interval (e.g., from 100ms to 5 seconds); when jobs appear, return to the minimum interval.

Monitoring the Worker Pool: Prometheus and Grafana

Without metrics, a worker pool is a black box. Key indicators to monitor: queue depth, active worker count, job processing time, error rate, and retry rate.

package metrics

import (
    "time"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    // Current job queue depth.
    QueueDepth = promauto.NewGauge(prometheus.GaugeOpts{
        Name: "worker_pool_queue_depth",
        Help: "Number of jobs waiting in the queue.",
    })

    // Number of active workers.
    ActiveWorkers = promauto.NewGauge(prometheus.GaugeOpts{
        Name: "worker_pool_active_workers",
        Help: "Number of currently active workers.",
    })

    // Histogram of job processing duration.
    JobDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "worker_pool_job_duration_seconds",
            Help:    "Time spent processing a job.",
            Buckets: prometheus.ExponentialBuckets(0.001, 2, 15), // 1ms..16s
        },
        []string{"job_type", "status"}, // status: success | error
    )

    // Counter of processed jobs.
    JobsTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "worker_pool_jobs_total",
            Help: "Total number of processed jobs.",
        },
        []string{"job_type", "status"},
    )

    // Retry counter.
    RetriesTotal = promauto.NewCounter(prometheus.CounterOpts{
        Name: "worker_pool_retries_total",
        Help: "Total number of job retries.",
    })
)

// RecordJobExecution instruments a job execution.
func RecordJobExecution(jobType string, start time.Time, err error) {
    duration := time.Since(start).Seconds()
    status := "success"
    if err != nil {
        status = "error"
    }
    JobDuration.WithLabelValues(jobType, status).Observe(duration)
    JobsTotal.WithLabelValues(jobType, status).Inc()
}

Recommended Grafana alerts:

worker_pool_queue_depth > 1000 for 5 minutes — indicates overload or stuck workers.
rate(worker_pool_jobs_total{status="error"}[5m]) / rate(worker_pool_jobs_total[5m]) > 0.05 — error rate above 5%.
histogram_quantile(0.99, worker_pool_job_duration_seconds) > 30 — p99 latency exceeds 30 seconds.

Comparison with Existing Solutions: When to Build Your Own

Before implementing a custom worker pool, it's worth evaluating existing solutions:

asynq — a Redis-based queue with a rich UI (Asynq Monitor), support for schedules, priorities, and unique jobs. An excellent choice if Redis is already in your stack and you need quick integration.
river — a PostgreSQL-native job queue from the authors of pgx. It uses LISTEN/NOTIFY instead of polling and supports transactional enqueue (a job is added atomically within a business transaction). Ideal if PostgreSQL is your only dependency.
machinery — a heavier solution supporting multiple brokers (Redis, AMQP, SQS), well suited for distributed workflows.

When it makes sense to build your own worker pool:

Jobs are in-memory and persistence is not required (e.g., a request processing pipeline).
You need custom prioritization or job routing logic.
Minimizing external dependencies is critical (embedded services, edge).
Off-the-shelf solutions are overly feature-rich and add unwanted complexity.

Practical Example: Async Image Processing Service

Let's put it all together in a realistic example: an HTTP service accepts image uploads, places processing tasks (resizing, format conversion) into a PostgreSQL queue, and a worker pool picks them up and processes them concurrently.

package main

import (
    "context"
    "encoding/json"
    "log/slog"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/prometheus/client_golang/prometheus/promhttp"
    "github.com/yourorg/metrics"
    "github.com/yourorg/pgqueue"
)

type ImagePayload struct {
    S3Key   string   `json:"s3_key"`
    Widths  []int    `json:"widths"`
    Formats []string `json:"formats"`
}

// processImageJob contains the business logic for processing a single image.
func processImageJob(ctx context.Context, payload json.RawMessage) error {
    var p ImagePayload
    if err := json.Unmarshal(payload, &p); err != nil {
        return err
    }
    // Here: download from S3, resize, save back.
    // Simulated for this example.
    slog.Info("processing image", "s3_key", p.S3Key, "widths", p.Widths)
    time.Sleep(100 * time.Millisecond) // simulate work
    return nil
}

func runWorkers(ctx context.Context, queue *pgqueue.Queue, workerCount int) {
    sem := make(chan struct{}, workerCount) // semaphore to cap concurrency

    for {
        select {
        case <-ctx.Done():
            // Wait for all semaphore slots to be released.
            for i := 0; i < workerCount; i++ {
                sem <- struct{}{}
            }
            return
        default:
        }

        job, err := queue.Dequeue(ctx)
        if err != nil {
            slog.Error("dequeue error", "err", err)
            time.Sleep(time.Second)
            continue
        }
        if job == nil {
            // Queue is empty — adaptive pause.
            time.Sleep(200 * time.Millisecond)
            continue
        }

        // Acquire a semaphore slot.
        sem <- struct{}{}
        metrics.ActiveWorkers.Inc()

        go func(j *pgqueue.PGJob) {
            defer func() {
                <-sem // release slot
                metrics.ActiveWorkers.Dec()
            }()

            start := time.Now()
            err := processImageJob(ctx, j.Payload)
            metrics.RecordJobExecution(j.Type, start, err)

            if err != nil {
                slog.Error("job failed", "job_id", j.ID, "err", err)
                if qErr := queue.Fail(ctx, j.ID, j.Attempts); qErr != nil {
                    slog.Error("failed to mark job as failed", "err", qErr)
                }
                return
            }
            if err := queue.Complete(ctx, j.ID); err != nil {
                slog.Error("failed to mark job complete", "err", err)
            }
        }(job)
    }
}

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
    defer stop()

    db, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
    if err != nil {
        slog.Error("db connect failed", "err", err)
        os.Exit(1)
    }
    defer db.Close()

    queue := pgqueue.NewQueue(db)

    // HTTP server: accepts uploads and serves metrics.
    mux := http.NewServeMux()
    mux.Handle("/metrics", promhttp.Handler())
    mux.HandleFunc("/upload", func(w http.ResponseWriter, r *http.Request) {
        // Simplified: encode the payload and enqueue it.
        payload, _ := json.Marshal(ImagePayload{
            S3Key:   "uploads/" + r.URL.Query().Get("file"),
            Widths:  []int{320, 640, 1280},
            Formats: []string{"webp", "jpg"},
        })
        if _, err := db.Exec(r.Context(),
            `INSERT INTO jobs (type, payload) VALUES ('image_process', $1)`, payload); err != nil {
            http.Error(w, "enqueue failed", http.StatusInternalServerError)
            return
        }
        w.WriteHeader(http.StatusAccepted)
    })

    srv := &http.Server{Addr: ":8080", Handler: mux}
    go srv.ListenAndServe()

    // Start the worker pool.
    go runWorkers(ctx, queue, 20)

    <-ctx.Done()
    slog.Info("shutting down")
    shutCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    srv.Shutdown(shutCtx)
}

This example demonstrates the full cycle: HTTP → PostgreSQL queue → worker pool with semaphore → Prometheus metrics → graceful shutdown. The semaphore (chan struct{}) replaces an explicit goroutine pool and is an idiomatic Go pattern for bounding concurrency.

Conclusion

A worker pool in Go is not just an optimization pattern — it's a fundamental building block for reliable, high-load services. We've covered the full spectrum: from a basic channel-based implementation to dynamic scaling, retry logic without goroutine leaks, graceful shutdown, and PostgreSQL as a persistent queue with FOR UPDATE SKIP LOCKED.

Key takeaways:

Always bound concurrency explicitly — via pool size, a semaphore, or a buffered channel.
Context must flow through the entire stack: from the HTTP handler down to the database call.
Graceful shutdown with a timeout is mandatory in production — the OS won't wait forever.
For persistent jobs, PostgreSQL with FOR UPDATE SKIP LOCKED is a mature and reliable solution, especially when paired with the river library.
Prometheus metrics should be built in from day one — don't add them as an afterthought.

Go provides all the primitives needed to build a production-ready worker pool without external dependencies. Mastering these patterns is what separates developers who can build systems that hold up under real-world load.

Go Worker Pool and Concurrent Task Processing: Patterns for High-Load Systems