Go worker pool

用固定数量 worker 消费任务 channel 控制并发度；所有权、收尾、取消与背压，以及与 pipeline 的分工边界。

#concept #growing #dev #go

[!info] related notes

所属 MOC：并发 MOC · 服务工程 · 学习路线

前置概念：goroutine、channel、select、context、sync

后续概念：pipeline、限流、优雅停机

关系总览：取消传播关系

容易混淆：无界 go 风暴、消息队列、无限缓冲 channel

Go worker pool

这个概念为什么会出现

go process(job) 写起来极便宜，但工程上会立刻撞到：

任务量远大于 CPU / 连接池 / 下游 QPS 能承受的并发
瞬时尖峰把进程打成成千上万 goroutine，调度与内存抖动
需要明确上限、可观测的在途任务数、可取消的收尾

worker pool 的答案是：预先（或按需但有上限地）启动固定数量的 worker，从共享任务队列取活。它不发明新语言特性，而是把“并发度控制 + 任务所有权 + 收尾”组合成稳定模式。官方在 Go Concurrency Patterns: Pipelines 里用 fan-out / fan-in 展示了同族思想：多 goroutine 从同一输入 channel 取任务，再汇总结果。

[!abstract] 一句话理解 worker pool = 有界 worker 集合 + 任务输入 channel（可选结果 channel）+ 明确的关闭/取消/等待协议；目标是控制并发资源，而不是“开更多 goroutine 就更快”。

最小可运行示例

先把示例放进业务场景，再看代码：

场景：批量缩略图任务，固定 3 个 worker

运营上传 5 张图，每张要做一次 CPU/IO 混合处理（这里用 j*j 假装“算缩略图”）。
若每张图 go process()，高峰会把机器打满。

worker pool：固定 3 个工人抢 jobs 队列；结果写入 results。
关闭纪律：

生产者关 jobs（没有新图了）
等所有 worker 退出后再关 results（不能让某个 worker 提前 close）

package main

import (
	"fmt"
	"sync"
)

func main() {
	// jobs：待处理图片 ID 队列（缓冲 8 = 有限排队，满则背压）
	jobs := make(chan int, 8)
	// results：处理结果（演示用平方值）
	results := make(chan int, 8)

	const workers = 3 // 并发上限：同时最多 3 张在处理
	var wg sync.WaitGroup
	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			// range jobs：队列关闭且排空后退出
			for jobID := range jobs {
				// 业务：生成缩略图；这里用平方假装 CPU 工作
				_ = workerID
				results <- jobID * jobID
			}
		}(w)
	}

	// 协调者：投递任务 → 关 jobs → 等工人 → 关 results
	go func() {
		for j := 1; j <= 5; j++ {
			jobs <- j // 5 张图
		}
		close(jobs) // 发送方关闭：工人的 range 才能结束
		wg.Wait()   // 所有工人不再写 results
		close(results)
	}()

	// 主流程：收齐全部结果（顺序不保证）
	for r := range results {
		fmt.Println("thumb result:", r)
	}
}

建议运行：

go run .

期望：打印 5 个结果，集合为 1 4 9 16 25（顺序可能打乱）。

结合场景再看三个关注点

worker 数限制池内同时执行的任务数；每个任务再开 goroutine、建立多条连接或持有大对象时，总资源并不直接等于 worker 数。
close(jobs) 只有生产者做；工人只消费。
close(results) 在所有发送者结束后由协调者做，避免重复关闭或向已关闭 channel 发送。主流程同时收结果，不能先阻塞在 Wait 再开始收，除非缓冲确实足以容纳全部结果。

核心概念与准确模型

组成部件

部件	职责
job channel	任务队列；缓冲大小决定排队深度与背压
worker 集合	固定 N 个循环：取任务 → 处理 → 可选写结果
result channel / 回调	输出路径；所有权必须单一
WaitGroup / errgroup	等待 worker 退出
context	发出协作取消信号；生产者、worker 与回源操作需观察它

与 fan-out 的关系

在 pipelines 博文中：

fan-out：多个函数从同一 channel 读（竞争消费）
fan-in：多个 channel 汇入一个

经典 worker pool 就是 fan-out 消费 jobs +（可选）fan-in 汇总 results。不必把“pool 对象”做成框架，channel 本身就是队列。

有界 vs 无界

无界：每个任务 go f(job)     → 并发度 = 任务数（危险）
有界：N workers + jobs 队列  → 并发度 ≤ N

jobs 缓冲满时会阻塞发送，但只约束 channel 内排队数量。若上游为每个发送另起 goroutine，或先把全部大任务读进内存，仍可能无界增长；结果缓冲和等待投递者也要纳入容量预算。

取消语义

// 策略：遇到任务错误就返回，由调用方（如 errgroup）收集。
func worker(ctx context.Context, jobs <-chan Job, out chan<- Result) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case j, ok := <-jobs:
			if !ok {
				return nil
			}
			// 处理时也要能被取消：优先用支持 ctx 的 API
			r, err := do(ctx, j)
			if err != nil {
				return err
			}
			select {
			case out <- r:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
	}
}

取消不是“杀死 goroutine”，而是协作退出。上述片段的 Job、Result、do 由业务定义；调用方必须处理返回错误，生产者的发送也要能被取消，否则 worker 退出后生产者可能永久阻塞。

Done 与任务/输出同时就绪时，select 可以选择任一分支，不能保证 cancel 后绝不取任务或发结果。要保证「停止返回后无结果」，需取消后等待所有发送者退出；资源清理必须由各任务配合。详见 go-context 与 go-goroutine-channel-context-and-cancellation-relationship。

错误与部分失败

常见策略：

errgroup.WithContext：首个非 nil 错误会取消派生 ctx，其他任务仍须协作退出；Wait 等待全部任务并返回首个错误。普通零值 Group 不自带取消，取消也不回滚已发生的副作用。
结果结构体带 error：适合“尽量完成，汇总失败项”。
独立 error channel：要严格所有权与关闭顺序，复杂度更高。

动态池？

Go 标准库没有 Java 式 ThreadPoolExecutor。多数服务用：

固定 N（=CPU 或下游限额）
或信号量（chan struct{} / semaphore）限制在途 goroutine

“动态扩缩”通常是运维层（副本数）问题，而不是进程内无限加 worker。

边界情况

jobs 不关闭：有限批次的 range 消费者会在无更多发送时一直等；长期服务中的等待可以是正常状态，但停机必须有关闭或取消协议。
results 被 worker 各自 close：可能重复关闭或其他 worker 向关闭的 channel 发送而 panic。
main 不等待 worker：进程退出截断未完成任务。
缓冲过大：伪装成无界队列，延迟暴露背压。
任务内再无限 go：池外泄漏，上限形同虚设。
共享可变状态：池只限并发数，不自动消除 data race（需 -race 与同步）。
取消后仍向 out 发送：接收方已退出 → 发送阻塞 → 二次泄漏。

常见误区

[!warning] 常见误区：worker pool 是默认并发方案错误：所有并发都套 pool。
正确：独立短任务且自然有界可用 errgroup；阶段流水线用 pipeline；共享结构用 mutex。

[!warning] 常见误区：N 越大越快错误：workers = 10000 处理 CPU 密集任务。
正确：CPU 密集常近 GOMAXPROCS；I/O 密集按下游配额与延迟实验定。

[!warning] 常见误区：关闭 = 取消错误：只 close jobs 就以为进行中的 HTTP 会停。
正确：关闭前先让所有生产者停止发送；range 会继续消费已缓冲任务再退出。取消可以选择放弃剩余任务，但不强制打断任意代码。

[!warning] 常见误区：把 pool 做成全局可变单例乱投递错误：无超时、无背压、无关闭协议的全局队列。
正确：生命周期绑定服务/请求范围，协议写进类型与文档。

工程实践

单向 channel 限制读写操作；关闭权仍由协议规定，chan<- Result 在语言层面也能被 close，不会自动保证唯一拥有者。
用 context 贯穿处理函数，defer cancel() 在创建点。
优雅停机先停止新任务来源，再明确选择排空或取消；等生产者停止后由唯一拥有者关闭 jobs，持续接收结果直到 worker 退出，再关闭 results。HTTP listener 通常在 Shutdown 开始时关闭，不能把它固定放在所有工作完成之后（go-graceful-shutdown）。
指标：队列长度、在途数、处理时延、失败率（go-observability）。
任务幂等：至少一次投递时下游要可重试安全。
测试：用小 N、可控 jobs、fake clock 或即时完成任务；加 -race。
优先标准组合：errgroup.Group + 信号量往往比自研“万能 Pool 结构体”更清晰。

// 信号量式有界并发（不必常驻 worker）
sem := make(chan struct{}, 8)
var wg sync.WaitGroup
dispatch:
for _, job := range all {
	select {
	case sem <- struct{}{}:
	case <-ctx.Done():
		break dispatch
	}
	wg.Add(1)
	go func(j Job) {
		defer wg.Done()
		defer func() { <-sem }()
		_ = handle(ctx, j)
	}(job)
}
wg.Wait()

此片段只演示取得许可后再启动 goroutine，取消投递后仍 Wait 已启动的任务；handle 的错误尚未汇总。若任务忽略 ctx，Wait 仍可能一直等待。select 的同时就绪分支仍没有取消优先级。

常驻 pool 常用于持续任务流，信号量适合限制批次扇出；并非两者的硬性适用边界。

可验证实验

实验 1：关闭顺序

去掉 close(jobs)，观察程序是否卡住。

实验 2：错误关闭 results

让每个 worker 处理完后 close(results)，观察 panic。

实验 3：背压

jobs 无缓冲、worker 处理很慢，看生产者是否在发送处阻塞。

实验 4：取消

让 worker 阻塞在取任务或无人接收的输出上，再 cancel 并等待 worker 返回。不要断言 cancel 调用瞬间之后没有一次就绪输出；只有等待所有发送者退出后才能保证不再发送。

实验 5：对比无界 go

对 1e5 任务分别“每任务一个 goroutine”与“8 worker”，比较峰值内存与完成时间（本机观察即可）。

本节总结

本质：有界并发的任务消费模式，不是魔法调度器。
关键规则：发送方关闭输入；协调者关闭输出；取消与关闭分工；等待所有 worker。
最易错：关闭权混乱、无取消的“假收尾”、无限缓冲伪装。
下一步：阶段化数据流见 go-pipeline-pattern；取消总图见 go-goroutine-channel-context-and-cancellation-relationship。

自测题

概念题

worker pool 主要限制的是什么资源行为？
谁应该 close(jobs)？谁应该 close(results)？
close jobs 与 cancel context 分别停止什么？

代码推理题

jobs := make(chan int)
go func() {
	jobs <- 1
	// 忘记 close(jobs)
}()
for j := range jobs {
	fmt.Println(j)
}

程序能否退出？为什么？

工程思考题

HTTP 服务要把上传文件扇出到 50 个第三方 API。如何选 worker 数、队列缓冲，以及客户端断开时如何停？

参考答案

展开

同时执行的任务数（及由此绑定的 CPU/连接/FD 等）。
jobs 由任务发送方/拥有者关闭；results 由“确认所有发送 results 的 goroutine 已结束”的协调者关闭。
close jobs：禁止后续发送，但缓冲中的任务仍可消费；cancel：发出协作退出信号，是否排空、丢弃或重试剩余任务取决于策略。代码题：不能正常结束，range jobs 在未关闭且无更多发送时阻塞（发送方已退出则死锁/泄漏）。
工程题：worker 上限对齐第三方配额与本机连接池；小缓冲制造背压；请求 ctx 取消时停止读队列并取消 in-flight HTTP；用 errgroup 或结果汇总错误。

延伸阅读与资料来源

资料	类型	支撑
Go Concurrency Patterns: Pipelines	官方博客	fan-out/fan-in、取消、阶段组合
Go Concurrency Patterns: Context	官方博客	取消在调用树中的传播
Share Memory By Communicating	官方博客	channel 表达并发结构
errgroup	扩展库	错误与取消组
Effective Go — Concurrency	文档	并发风格

笔记元信息

建议文件名：go-worker-pool.md
所属阶段：并发模式
建议下一篇：Go pipeline 模式
本篇状态：已深化（结构完整；对齐 pipelines 博文术语）