First, clarify the answer: The core of implementing a work pool in Go is to use a fixed number of goroutines to read and process tasks from the task channel, ensure that all tasks are completed through WaitGroup, and the channel is closed reasonably to avoid leakage. The specific steps are: 1. Define the Job and Result structures for passing tasks and results; 2. Create buffered task channels and result channels; 3. Start a fixed number of work goroutines, and each worker reads tasks from the task channel and processes them; 4. Use sync.WaitGroup to wait for all workers to complete the work; 5. Close the task channel after the main coroutine sends the task; 6. Close the result channel after all workers are finished; 7. The main coroutine collects results until the result channel is closed; 8. Cancel mechanism can be implemented through context to enhance control capabilities. This mode is suitable for processing a large number of similar tasks, can effectively control the use of resources and prevent the surge in goroutines. It is often used in batch processing, current limiting and background task systems in production environments, ultimately ensuring efficient and stable operation of the program.

Implementing a worker pool in Go is a common and effective way to manage concurrent tasks efficiently, especially when dealing with a large number of jobs without overwhelming system resources. Instead of spawning a goroutine for every task (which can lead to high memory usage and context-switching overhead), a worker pool limits concurrency by reusing a fixed number of workers to process jobs from a shared queue.

Here's how to implement a simple yet practical worker pool in Go.

Understanding the Worker Pool Pattern

A worker pool consists of:

• A fixed number of workers (goroutines)
• A job queue (channel) that holds incoming tasks
• A result queue (optional channel) for collecting results
• A way to signal completion (often using sync.WaitGroup )

Workers continuously pull jobs from the job channel and process them until the channel is closed.

Basic Implementation

 package main

import (
    "fmt"
    "sync"
    "time"
)

// Job represents a task to be processed
type Job struct {
    ID int
    Data string
}

// Results represent the output of a job
type Result struct {
    JobID int
    Success bool
    Msg string
}

// Worker function
func worker(id int, jobs <-chan Job, results chan<- Results, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        // Simulate work
        time.Sleep(500 * time.Millisecond)
        fmt.Printf("Worker %d processing job %d: %s\n", id, job.ID, job.Data)

        // Simulate success/failure
        var success bool
        if job.ID%2 == 0 {
            success = true
        } else {
            success = false
        }

        // Send result
        results <- Results{
            JobID: job.ID,
            Success: success,
            Msg: fmt.Sprintf("Job %d completed", job.ID),
        }
    }
}

// StartWorkerPool initializes the pool
func StartWorkerPool(numWorkers int, jobs <-chan Job, results chan<-Result) {
    var wg sync.WaitGroup

    // Start workers
    for i := 1; i <= numWorkers; i {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // Wait for all workers to finish
    go func() {
        wg.Wait()
        close(results) // Signal no more results
    }()
}

Using the Worker Pool

 func main() {
    numJobs := 10
    numWorkers := 3

    jobs := make(chan Job, numJobs)
    results := make(chan Results, numJobs)

    // Send jobs
    go func() {
        for i := 1; i <= numJobs; i {
            jobs <- Job{ID: i, Data: fmt.Sprintf("payload-%d", i)}
        }
        close(jobs) // Important: close job channel to stop workers
    }()

    // Start pool
    StartWorkerPool(numWorkers, jobs, results)

    // Collect results
    for result := range results {
        status := "SUCCESS"
        if !result.Success {
            status = "FAILED"
        }
        fmt.Printf("Result: Job %d | %s | %s\n", result.JobID, status, result.Msg)
    }

    fmt.Println("All jobs processed.")
}

Key Design Points

• Buffered job channel : Allows decoupling of job submission and processing.
• Closing the job channel : Tells workers there are no more jobs. Using range on a closed channel drains it and exits cleanly.
• WaitGroup in worker pool : Ensures all workers finish before closing the results channel.
• Results channel : Optional. Use it if you need feedback (eg, success/failure, processed data).
• No goroutine leaks : Proper channel closure ensures workers exit and don't block forever.

When to Use a Worker Pool

• Processing thousands of similar tasks (eg, file parsing, HTTP requests, DB inserts)
• Rate limiting or controlling resource usage
• Background job processing
• Avoiding unbounded goroutine creation

Possible Enhancements

You can extend this pattern with:

• Error handling and retries
• Context cancellation for graceful shutdown
• Dynamic worker scaling (advanced)
• Prioritized job queues using multiple channels or select

Example with context:

 func workerWithContext(ctx context.Context, id int, jobs <-chan Job, results chan<-Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case job, ok := <-jobs:
            if !ok {
                return // Channel closed
            }
            // Process job
            time.Sleep(200 * time.Millisecond)
            results <- Result{JobID: job.ID, Success: true}
        case <-ctx.Done():
            fmt.Printf("Worker %d shutting down...\n", id)
            Return
        }
    }
}

This pattern gives you control, visibility, and efficiency. It's widely used in production Go services for batch processing, API rate limiting, and background workers.

Basically, keep it simple: fixed workers, a job channel, and proper cleanup .

The above is the detailed content of Implementing a Worker Pool in Go. For more information, please follow other related articles on the PHP Chinese website!