wonder/cs-note

Fork 0

Files

T

wonder 42f0507747 vault backup: 2026-05-31 17:40:37

2026-05-31 17:40:37 +08:00

11 KiB

Raw Permalink Blame History

tags, create time

Go 协程池

概述

梳理 Go 中协程池（Goroutine Pool）的设计动机与两种主流实现方案——基于 channel 的轻量级池子和 golang.ccpool 风格的成熟方案，分析其核心模式、线程安全保证以及适用边界。

正文

一、为什么需要协程池

[!question] 思考：如果一个服务每秒收到 10000 个请求，每个请求启动一个 goroutine，会发生什么？

Go 的 goroutine 极其轻量——初始栈仅 2KB，创建成本约几个纳秒——所以 Go 社区流行"别管它，随时起一个"的信条。但轻量不等于没有成本。goroutine 仍然占用内存、参与 GC scan、需要调度器时间片。不加控制的 goroutine 爆炸在真实系统中屡见不鲜：

// ❌ 灾难写法
for _, item := range items {
    go process(item) // 如果 items 有 100 万个元素……
}
time.Sleep(time.Hour) // 等所有完成——但这也不是可靠的做法

当 goroutine 数量暴增时会导致：

内存暴涨：即便每个只有 2KB，百万级就是 GB 级别
GC 压力：大量短命对象触发频繁 Full GC
CPU 抖动：太多 runnable goroutine 导致调度竞争

策略	控制粒度	实现复杂度	适用场景
直接并发	无	最低	小批量、短时间任务
Worker 池	goroutine 数量	中等	中等并发、需限流
Semaphore (channel)	最大并发数	低	快速限速

二、Channel + WaitGroup：最简协程池

这是最经典的实现方式——用一个固定数量的 worker 集合加一个任务 channel：

func Worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs { // range 自动退出——jobs channel 关闭后结束
        fmt.Printf("Worker %d started job %d\n", id, j)
        time.Sleep(time.Second)
        results <- j * 2
    }
}

func main() {
    const numWorkers = 3
    jobs := make(chan int, 100)       // 带缓冲的任务通道
    results := make(chan int, 100)     // 结果通道

    var wg sync.WaitGroup
    for w := 0; w < numWorkers; w++ {
        wg.Add(1)
        go Worker(w, jobs, results, &wg)
    }

    // 投递任务
    for j := 1; j <= 10; j++ {
        jobs <- j
    }
    close(jobs)  // 关键：关闭 channel，worker 的 range 才能退出

    go func() {
        wg.Wait()
        close(results)
    }()

    // 消费结果
    for r := range results {
        fmt.Println("result:", r)
    }
}

这个模式的精妙之处在于用 channel 的语义替代了锁：

range jobs 天然处理了"取任务 → 执行 → 取下一个"的循环
close(jobs) 作为停止信号，worker 不用轮询或检查 cancel
WaitGroup 确保所有 worker 完成后再关闭 results

[!tip] Channel 缓冲大小的选择

太大：失去限速效果，退化成全并发

太小：worker 频繁阻塞在取任务上，吞吐量下降

经验值：等于或略大于 worker 数量（如 1-3 倍）即可，除非你确信生产者远快于消费者

2.1 结构示意

flowchart LR
    subgraph Producer["生产者"]
        P["task 1, 2, 3..."]
    end

    subgraph Pool["协程池"]
        J["jobs chan\n(buffered)"]
        W1["Worker 1"]
        W2["Worker 2"]
        W3["Worker 3"]
        J --> W1
        J --> W2
        J --> W3
    end

    R["results chan"]
    W1 --> R
    W2 --> R
    W3 --> R

    P --> J
    R --> Consumer["消费者"]

    style Pool fill:#e8f0fe,stroke:#1a73e8
    style J fill:#fff3e0,stroke:#f9a825

三、函数式选项：更友好的 API 封装

实际项目中通常会把上面的 boilerplate 封装成一个易用的函数：

// GoPool 是一个简单的协程池
type GoPool struct {
    workers int
    queue   chan func()
}

// NewGoPool 创建一个指定 worker 数量的协程池
func NewGoPool(workers int) *GoPool {
    return &GoPool{
        workers: workers,
        queue:   make(chan func(), 1000), // 队列缓冲
    }
}

// Start 启动 worker，返回 shutdown 函数
func (p *GoPool) Start(ctx context.Context) {
    var wg sync.WaitGroup
    for i := 0; i < p.workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for {
                select {
                case fn, ok := <-p.queue:
                    if !ok {
                        return // channel 关闭，退出
                    }
                    fn()
                case <-ctx.Done(): // 支持外部取消
                    return
                }
            }
        }()
    }

    // 返回 shutdown 回调：关闭 channel + 等待 worker 退出
    p.shutdown = func() {
        close(p.queue)
        wg.Wait()
    }
}

使用变得极其简洁：

pool := NewGoPool(10)
pool.Start(context.Background())

// 提交异步任务
pool.Submit(func() {
    doSomething()
})

// 清理
pool.Shutdown() // close(channel) + WaitGroup.Wait()

[!note] Submit 之后如何拿到结果？

上面的版本是 fire-and-forget 模式（不返回值）。如果需要收集结果，让匿名函数写入一个共享 channel，或者用 sync.Once / errgroup.Group 做同步等待。这也是为什么下面要引入更成熟的工具。

四、errgroup：官方级别的简化方案

[!important] 先看这里

golang.org/x/sync/errgroup 虽然不是传统意义上的"池"，但它解决了最常见的并发痛点：有限度并发 + 错误传播 + 取消传递。大多数情况下不需要自己写池子。

func downloadUrls(urls []string) error {
    g, ctx := errgroup.WithContext(context.Background())
    g.SetLimit(10) // ⚡ 限制最多 10 个并发 goroutine

    for _, url := range urls {
        u := url // 注意循环变量捕获陷阱
        g.Go(func() error {
            select {
            case <-ctx.Done(): // 任意一个失败，全部取消
                return ctx.Err()
            default:
                return fetch(u)
            }
        })
    }

    return g.Wait() // 等全部完成，或第一个错误返回
}

errgroup 的核心价值：

SetLimit(10)：一行代码搞定并发限制，本质上是 channel semaphore
Go(fn) 返回 error：任何一个 goroutine panic 或返回非 nil error，整个组取消并提前退出
WithCancel + WaitGroup 一体化：不用分别管理

[!summary] 选哪个方案？

flowchart TD
    A["需要限制并发数？"] -->|否| B["直接 go + 不管"]
    A -->|是| C["errgroup.SetLimit()"]
    C --> D["还需要自定义任务类型\n(结构化数据流)?"]
    D -->|否| E["✅ errgroup 够用了"]
    D -->|是| F["自定义协程池\nchannel + Worker"]
    F --> G["还需要优雅停机?\n(关闭后拒绝新任务)?"]
    G -->|是| H["golang.ccpool /\ntpool 等第三方库"]
    G -->|否| E

五、Semaphore：另一种限速思路

除了 errgroup 内部用的 buffered channel + counter 方案，还有一种更直观的 semaphore 写法：

type semaphore struct {
    ch chan struct{}
}

func newSemaphore(n int) semaphore {
    return semaphore{ch: make(chan struct{}, n)}
}

func (s semaphore) acquire()  { s.ch <- struct{}{} }
func (s semaphore) release()  { <-s.ch }

配合 goroutine 使用的模板：

sem := newSemaphore(10)

for _, item := range items {
    go func(i Item) {
        sem.acquire()
        defer sem.release()
        process(i)
    }(item)
}

这和 errgroup 的 SetLimit 底层原理相同，只是手动暴露了 acquire/release。适合需要将 semaphores 传给多个独立函数的场景。

六、生产级协程池的特性要求

如果要在真实产品中落地自己的池子，通常需要考虑以下能力：

特性	说明	实现要点
优雅停机	Stop 后不接收新任务，等待已提交的完成	state machine (running/stopping/stopped)，Select 判断状态
动态扩缩容	根据负载调整 worker 数量	monitor 定时检查队列长度
任务超时	单个任务执行超时自动跳过	`select` + `time.After`
优先级队列	高优先级任务先执行	多 channel 或 heap 结构
监控指标	活跃 worker、排队数、丢弃数	atomic counters + prometheus registry

下面是优雅停机的核心逻辑骨架：

type State int

const (
    StateRunning State = iota
    StateStopping
    StateStopped
)

type SafePool struct {
    state   State
    mu      sync.RWMutex
    queue   chan Task
    workers int
    done    chan struct{}
}

func (p *SafePool) Submit(t Task) error {
    p.mu.RLock()
    defer p.mu.RUnlock()

    if p.state == StateStopped {
        return ErrPoolStopped
    }

    select {
    case p.queue <- t:
        return nil
    default:
        return ErrQueueFull // 队列满时非阻塞拒绝
    }
}

func (p *SafePool) Stop() {
    p.mu.Lock()
    defer p.mu.Unlock()

    if p.state == StateStopped {
        return
    }
    p.state = StateStopping
    close(p.done)   // 通知 worker 退出
    p.state = StateStopped
}

七、常见陷阱清单

7.1 goroutine 泄漏

// ❌ worker 永远阻塞在有缓冲的 channel 上，如果没人读
jobs := make(chan int, 10) // 缓冲 10
jobs <- 1
// 如果 worker 没有启动，或者 worker 全部退出了但 channel 里还有未取的数据……
// goroutine 就泄漏了！

// ✅ 始终用 WaitGroup 或 context 跟踪所有 goroutine 的生命周期

7.2 循环变量捕获（Go 1.22 之前）

// ❌ Go 1.22 之前的版本，所有 goroutine 共享同一个 i
for i := 0; i < 5; i++ {
    go func() {
        fmt.Println(i) // 可能全部输出 5！
    }()
}

// ✅ 显式传参（推荐）或就地拷贝
for i := 0; i < 5; i++ {
    go func(val int) {
        fmt.Println(val)
    }(i)
}

[!note] Go 1.22 变更

从 Go 1.22 开始，循环变量的行为被修改为每次迭代创建新的副本。如果你的项目仍跑在更早的版本上，上述问题依然是真实的 bug。

7.3 channel 未关闭导致阻塞

// ❌ main 函数在 wg.Wait() 之前就返回了——results 永远不会被关闭
var wg sync.WaitGroup
for w := 0; w < 3; w++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for j := range jobs {
            results <- j * 2
        }
    }()
}
wg.Wait()
// 此处缺少 close(results)

// ✅ 独立的 goroutine 负责在 wg.Wait() 后关闭 results
go func() {
    wg.Wait()
    close(results)
}()

7.4 队列满了之后无限阻塞

// ❌ submit 会一直阻塞直到有人消费
p.queue <- task

// ✅ 用 select + timeout/done channel 避免死锁
select {
case p.queue <- task:
    return nil
case <-ctx.Done():
    return ctx.Err()
default:
    return ErrQueueFull
}

关联笔记

Go Context

11 KiB Raw Permalink Blame History Unescape Escape