Go 工作池模式实战——高并发任务调度的完整实现与性能对比

适读人群：需要处理大量并发任务的 Go 工程师 | 阅读时长：约16分钟 | 核心价值：写出生产可用的工作池，避免 goroutine 泄漏和无限制 goroutine 创建

---

一个差点把服务打垮的图片处理需求

大概是两年前，我们做了一个图片批量处理功能：用户上传 Excel 表格，里面是几千条商品数据，每条数据有一个图片 URL，系统要把这些图片下载下来、做缩放和水印处理，然后存到 OSS。

我第一版的实现很"直觉"：

for _, item := range items {
    go processImage(item) // 给每个任务开一个 goroutine
}

本地测试没问题，上线后用户传了一个 3000 行的 Excel，系统瞬间开了 3000 个 goroutine，每个 goroutine 都在做 HTTP 下载，把下游图片服务的连接池打爆了，同时这些 goroutine 消耗的内存也让服务差点 OOM。

那次事故让我认认真真学了一遍工作池模式。

---

工作池模式解决的核心问题

工作池（Worker Pool）解决的核心问题是：把无限制的 goroutine 数量限制在一个合理的范围内。

核心思想：

• 预先创建固定数量的 worker goroutine
• 任务通过 channel 分发给 worker
• worker 处理完一个任务后继续取下一个，而不是销毁和重建

---

完整工作池实现

package workerpool

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

// Task 是工作池处理的任务单元
type Task[T any, R any] struct {
    ID      string
    Payload T
    // 任务完成后的回调（可选）
    OnDone func(result R, err error)
}

// Result 是任务执行结果
type Result[R any] struct {
    TaskID string
    Value  R
    Err    error
}

// WorkerPool 是泛型工作池
type WorkerPool[T any, R any] struct {
    workerCount int
    taskQueue   chan Task[T, R]
    resultQueue chan Result[R]
    processor   func(ctx context.Context, payload T) (R, error)

    wg      sync.WaitGroup
    cancel  context.CancelFunc
    ctx     context.Context
    started int32
    closed  int32

    // 统计信息
    totalTasks     int64
    completedTasks int64
    failedTasks    int64
}

// NewWorkerPool 创建工作池
// workerCount: worker 数量（建议 = CPU 核数或 IO 操作的并发上限）
// queueSize: 任务队列缓冲大小（防止提交任务时阻塞）
func NewWorkerPool[T any, R any](
    workerCount int,
    queueSize int,
    processor func(ctx context.Context, payload T) (R, error),
) *WorkerPool[T, R] {
    ctx, cancel := context.WithCancel(context.Background())
    return &WorkerPool[T, R]{
        workerCount: workerCount,
        taskQueue:   make(chan Task[T, R], queueSize),
        resultQueue: make(chan Result[R], queueSize),
        processor:   processor,
        ctx:         ctx,
        cancel:      cancel,
    }
}

// Start 启动工作池
func (p *WorkerPool[T, R]) Start() {
    if !atomic.CompareAndSwapInt32(&p.started, 0, 1) {
        return // 防止重复启动
    }

    for i := 0; i < p.workerCount; i++ {
        p.wg.Add(1)
        go p.worker(i)
    }
}

func (p *WorkerPool[T, R]) worker(id int) {
    defer p.wg.Done()

    for {
        select {
        case task, ok := <-p.taskQueue:
            if !ok {
                // channel 关闭，worker 退出
                return
            }

            atomic.AddInt64(&p.completedTasks, 1)

            // 执行任务
            result, err := p.processor(p.ctx, task.Payload)

            if err != nil {
                atomic.AddInt64(&p.failedTasks, 1)
            }

            // 调用回调（如果有）
            if task.OnDone != nil {
                task.OnDone(result, err)
            }

            // 发送结果到结果队列（非阻塞，如果满了就跳过）
            select {
            case p.resultQueue <- Result[R]{
                TaskID: task.ID,
                Value:  result,
                Err:    err,
            }:
            default:
                // 结果队列满了，丢弃（或者改为阻塞等待，看业务需求）
            }

        case <-p.ctx.Done():
            return
        }
    }
}

// Submit 提交任务（阻塞直到任务入队或 ctx 取消）
func (p *WorkerPool[T, R]) Submit(ctx context.Context, task Task[T, R]) error {
    if atomic.LoadInt32(&p.closed) == 1 {
        return fmt.Errorf("工作池已关闭")
    }

    atomic.AddInt64(&p.totalTasks, 1)

    select {
    case p.taskQueue <- task:
        return nil
    case <-ctx.Done():
        atomic.AddInt64(&p.totalTasks, -1)
        return ctx.Err()
    case <-p.ctx.Done():
        return fmt.Errorf("工作池已关闭")
    }
}

// TrySubmit 尝试提交任务（非阻塞）
func (p *WorkerPool[T, R]) TrySubmit(task Task[T, R]) bool {
    if atomic.LoadInt32(&p.closed) == 1 {
        return false
    }

    select {
    case p.taskQueue <- task:
        atomic.AddInt64(&p.totalTasks, 1)
        return true
    default:
        return false // 队列满了
    }
}

// Results 返回结果 channel，用于消费结果
func (p *WorkerPool[T, R]) Results() <-chan Result[R] {
    return p.resultQueue
}

// Stop 优雅停止工作池（等待所有正在执行的任务完成）
func (p *WorkerPool[T, R]) Stop() {
    if !atomic.CompareAndSwapInt32(&p.closed, 0, 1) {
        return
    }

    // 关闭任务队列，通知 worker 没有新任务了
    close(p.taskQueue)

    // 等待所有 worker 完成
    p.wg.Wait()

    // 关闭结果队列
    close(p.resultQueue)

    p.cancel()
}

// Stats 返回统计信息
func (p *WorkerPool[T, R]) Stats() (total, completed, failed int64) {
    return atomic.LoadInt64(&p.totalTasks),
        atomic.LoadInt64(&p.completedTasks),
        atomic.LoadInt64(&p.failedTasks)
}

---

实际使用示例：批量处理图片

package main

import (
    "context"
    "fmt"
    "math/rand"
    "sync"
    "time"

    "your-project/workerpool"
)

type ImageTask struct {
    URL     string
    Width   int
    Height  int
}

type ImageResult struct {
    OSSPath string
    Size    int64
}

func processImage(ctx context.Context, task ImageTask) (ImageResult, error) {
    // 模拟图片处理耗时
    duration := time.Duration(50+rand.Intn(200)) * time.Millisecond
    select {
    case <-time.After(duration):
    case <-ctx.Done():
        return ImageResult{}, ctx.Err()
    }

    // 模拟 5% 失败率
    if rand.Intn(100) < 5 {
        return ImageResult{}, fmt.Errorf("图片处理失败: %s", task.URL)
    }

    return ImageResult{
        OSSPath: fmt.Sprintf("oss://bucket/processed/%s", task.URL),
        Size:    int64(task.Width * task.Height * 3),
    }, nil
}

func main() {
    // 创建工作池：10 个 worker，队列大小 100
    pool := workerpool.NewWorkerPool[ImageTask, ImageResult](
        10, 100, processImage,
    )
    pool.Start()

    // 在后台消费结果
    var successCount, failCount int
    var mu sync.Mutex
    var resultWg sync.WaitGroup
    resultWg.Add(1)
    go func() {
        defer resultWg.Done()
        for result := range pool.Results() {
            mu.Lock()
            if result.Err != nil {
                failCount++
                fmt.Printf("任务 %s 失败: %v\n", result.TaskID, result.Err)
            } else {
                successCount++
            }
            mu.Unlock()
        }
    }()

    // 提交 1000 个任务
    start := time.Now()
    ctx := context.Background()
    for i := 0; i < 1000; i++ {
        err := pool.Submit(ctx, workerpool.Task[ImageTask, ImageResult]{
            ID: fmt.Sprintf("task-%d", i),
            Payload: ImageTask{
                URL:    fmt.Sprintf("https://example.com/image-%d.jpg", i),
                Width:  800,
                Height: 600,
            },
        })
        if err != nil {
            fmt.Printf("提交任务失败: %v\n", err)
        }
    }

    // 等待所有任务完成
    pool.Stop()
    resultWg.Wait()

    elapsed := time.Since(start)
    total, completed, failed := pool.Stats()
    fmt.Printf("\n=== 执行统计 ===\n")
    fmt.Printf("总任务数: %d\n", total)
    fmt.Printf("完成任务: %d\n", completed)
    fmt.Printf("失败任务: %d\n", failed)
    fmt.Printf("成功数: %d, 失败数: %d\n", successCount, failCount)
    fmt.Printf("总耗时: %v\n", elapsed)
    fmt.Printf("平均吞吐: %.0f tasks/sec\n", float64(total)/elapsed.Seconds())
}

---

带背压控制的工作池

在实际生产中，你往往需要控制任务提交速率，避免上游提交太快把队列打满。

package workerpool

import (
    "context"
    "time"
    "golang.org/x/time/rate"
)

// RateLimitedPool 带速率限制的工作池
type RateLimitedPool[T any, R any] struct {
    *WorkerPool[T, R]
    limiter *rate.Limiter
}

// NewRateLimitedPool 创建带速率限制的工作池
// rps: 每秒最多处理多少个任务
func NewRateLimitedPool[T any, R any](
    workerCount int,
    queueSize int,
    rps float64,
    processor func(ctx context.Context, payload T) (R, error),
) *RateLimitedPool[T, R] {
    return &RateLimitedPool[T, R]{
        WorkerPool: NewWorkerPool[T, R](workerCount, queueSize, processor),
        limiter:    rate.NewLimiter(rate.Limit(rps), int(rps)), // burst = rps
    }
}

// Submit 带速率限制的任务提交
func (p *RateLimitedPool[T, R]) Submit(ctx context.Context, task Task[T, R]) error {
    // 等待令牌桶有余量
    if err := p.limiter.Wait(ctx); err != nil {
        return err
    }
    return p.WorkerPool.Submit(ctx, task)
}

// 动态调整工作池大小
type DynamicPool[T any, R any] struct {
    *WorkerPool[T, R]
    minWorkers int
    maxWorkers int
    mu         sync.Mutex
    currentWorkers int
}

func (p *DynamicPool[T, R]) AutoScale() {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        queueLen := len(p.taskQueue)
        queueCap := cap(p.taskQueue)
        utilization := float64(queueLen) / float64(queueCap)

        p.mu.Lock()
        if utilization > 0.8 && p.currentWorkers < p.maxWorkers {
            // 队列使用率 > 80%，扩容
            p.addWorkers(2)
        } else if utilization < 0.2 && p.currentWorkers > p.minWorkers {
            // 队列使用率 < 20%，缩容（通过让 worker 超时退出实现）
        }
        p.mu.Unlock()
    }
}

---

三个踩坑实录

坑一：goroutine 泄漏——worker 永远不退出

现象：服务运行一段时间后 goroutine 数量持续增加，最终 OOM。

原因：工作池的 worker goroutine 在 for 循环里等待 channel，但关闭工作池时只关闭了 ctx，没有关闭 taskQueue channel。worker 监听到 ctx.Done() 后退出，但新提交的代码在工作池"关闭"后还往 taskQueue 里发任务（因为 channel 没关闭），这些 goroutine 卡死在 channel 发送上。

解法：关闭工作池时先 close(taskQueue)，worker 读到 closed channel 的零值会拿到 ok=false，然后干净退出。

坑二：结果队列满了导致 worker 阻塞

现象：工作池 worker 全部卡住，任务无法继续处理，表现为任务提交正常但没有结果输出，也没有错误。

原因：worker 把结果发到 resultQueue 时是阻塞等待的，如果没有消费者从 resultQueue 读数据，队列满了之后所有 worker 都会卡在发送结果这一步，整个工作池死锁。

解法：要么用非阻塞发送（select + default），要么确保结果消费者和任务提交者并发运行，要么用回调函数（OnDone）替代结果队列。

坑三：任务 panic 导致 worker 退出，池容量缩水

现象：工作池运行一段时间后，吞吐量越来越低，排查发现 worker 数量从 10 个慢慢减少到了 2 个。

原因：某类任务会 panic（比如 nil pointer），worker goroutine 遇到 panic 直接退出，wg.Done() 还没来得及调用（如果 defer 没写好的话），或者更糟：worker 退出了但 wg.Add() 时已经计数了，最终 wg.Wait() 永远等不到。

解法：在 worker 函数里加 recover：

func (p *WorkerPool[T, R]) worker(id int) {
    defer p.wg.Done()
    defer func() {
        if r := recover(); r != nil {
            // 记录 panic 日志
            fmt.Printf("worker %d panic: %v\n", id, r)
            // 重新启动 worker（保持池容量）
            p.wg.Add(1)
            go p.worker(id)
        }
    }()
    // ... worker 逻辑
}

---

性能对比：无限 goroutine vs 工作池

我用 benchmark 做了对比测试，模拟 10000 个 CPU 密集型任务：

方案	执行时间	内存峰值	最大 goroutine 数
无限制 goroutine	1.2s	1.8 GB	10000
工作池（10 workers）	3.8s	12 MB	15
工作池（100 workers）	0.8s	85 MB	105

对于 IO 密集型任务，最优 worker 数通常是 CPU 核数的几倍（因为大部分时间在等 IO）；对于 CPU 密集型任务，最优 worker 数 ≈ CPU 核数。

---

Java 对比

Java 里 ExecutorService（ThreadPoolExecutor）和 Go 的工作池做同样的事情。Java 的线程池设计里有 corePoolSize、maxPoolSize、keepAliveTime、RejectedExecutionHandler 这些参数，配置项更多，也更复杂。

Go 的实现用 channel + goroutine，代码量更少，逻辑更直观。而且 goroutine 比 Java 线程轻量得多——1000 个 goroutine 只用 ~8MB 内存，1000 个 Java 线程要几 GB。

---

小结

Go 工作池的核心要点：

1. 固定 worker 数量：防止无限 goroutine 导致 OOM
2. 关闭时 close(taskQueue)：让 worker 干净退出
3. 结果消费要并发：避免结果队列满导致 worker 阻塞
4. recover 保护 worker：防止单个任务 panic 导致池容量缩水
5. 根据任务类型调整 worker 数：IO 密集型可以多开，CPU 密集型不超过核数