Go Worker Pool 模式：用带缓冲 channel 把任务批处理与限流做稳

为什么需要 Worker Pool

每来一个任务就开一个 goroutine，在突发流量下 goroutine 数量会失控，内存与调度都吃紧。Worker Pool 用固定数量的 worker 常驻消费任务队列，既限制并发上限，又复用 goroutine，天然带限流效果。

基础结构

type Job struct {
    ID   int
    Data string
}

type Result struct {
    JobID  int
    Output string
    Err    error
}

type Pool struct {
    jobs    chan Job
    results chan Result
    workers int
}

启动 worker

每个 worker 从 jobs 通道循环取任务，处理完写回 results，通道关闭时退出：

func (p *Pool) worker() {
    for job := range p.jobs {
        out, err := process(job)
        p.results <- Result{JobID: job.ID, Output: out, Err: err}
    }
}

func process(j Job) (string, error) {
    // 实际业务逻辑，例如调用下游服务
    return "done-" + j.Data, nil
}

派发与收集

主流程把任务投进 jobs，再从 results 收齐，用 sync.WaitGroup 等所有 worker 退出：

func (p *Pool) Run(tasks []Job) []Result {
    var wg sync.WaitGroup
    for i := 0; i < p.workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            p.worker()
        }()
    }

    go func() {
        for _, t := range tasks {
            p.jobs <- t
        }
        close(p.jobs)
    }()

    var results []Result
    for r := range p.results {
        results = append(results, r)
    }

    wg.Wait()
    return results
}

用 context 控制取消

任务处理里监听 ctx.Done，上游超时时及时退出，避免资源空耗：

func process(ctx context.Context, j Job) (string, error) {
    select {
    case <-ctx.Done():
        return "", ctx.Err()
    case <-time.After(50 * time.Millisecond):
        return "done-" + j.Data, nil
    }
}

工程落地的几点

• worker 数不是越大越好，通常贴近 CPU 核数或下游连接池上限。
• results 通道要带缓冲，否则 worker 写回时会阻塞在派发协程上。
• 需要限流时，把 jobs 投送速率用 time.Tick 控住即可，pool 本身不关心来源。