409 lines
11 KiB
Markdown
409 lines
11 KiB
Markdown
---
|
||
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]]
|