---
tags: [go, goroutine, concurrency, pool]
create time: 2026-05-30 14:20
---

# Go 协程池

## 概述

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

## 正文

### 一、为什么需要协程池

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

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

```go
// ❌ 灾难写法
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：

```go
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 结构示意

```mermaid
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 封装成一个易用的函数：

```go
// 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()
    }
}
```

使用变得极其简洁：

```go
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` 虽然不是传统意义上的"池"，但它解决了最常见的并发痛点：**有限度并发 + 错误传播 + 取消传递**。大多数情况下不需要自己写池子。

```go
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] 选哪个方案？

```mermaid
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 写法：

```go
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 使用的模板：

```go
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 |

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

```go
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 泄漏

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

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

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

```go
// ❌ 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 未关闭导致阻塞

```go
// ❌ 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 队列满了之后无限阻塞

```go
// ❌ 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]]