学完这篇你能得到三个可以反复套用的并发骨架——Worker Pool、Fan-out、Fan-in 的标准写法，以及它们组合起来处理批量任务时的完整代码，直接复制就能跑。
第一步：先分清三个模式各管什么
- Worker Pool（工作池）：固定 N 个 goroutine 从同一个任务 channel 取活干。作用是限制并发数，避免一万个任务开一万个 goroutine 打爆内存或下游服务。
- Fan-out（扇出）：把同一批任务分发给多个处理者，本质就是 Worker Pool 的「启动多个 goroutine 消费同一个 channel」这个动作。
- Fan-in（扇入）：把多个 channel 的输出合并到一个 channel。作用是统一汇总结果，让调用方只 range 一个 channel。
三者关系：Fan-out 是分发，Fan-in 是回收，Worker Pool 是这两件事加起来的完整形态。
先建目录跑通环境：
mkdir goconc && cd goconc
go mod init goconc
第二步：写一个最小 Worker Pool
新建 main.go，核心是「生产者关 channel，消费者 range channel」：
package main
import (
"fmt"
"sync"
)
func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
defer wg.Done()
for j := range jobs {
results <- j * j
}
}
func main() {
const numJobs, numWorkers = 10, 4
jobs := make(chan int, numJobs)
results := make(chan int, numJobs)
var wg sync.WaitGroup
for w := 1; w <= numWorkers; w++ {
wg.Add(1)
go worker(w, jobs, results, &wg)
}
go func() {
wg.Wait()
close(results) // 所有 worker 退出后才关结果 channel
}()
for i := 1; i <= numJobs; i++ {
jobs <- i
}
close(jobs) // 任务发完，通知 worker 退出 range
for r := range results {
fmt.Println(r)
}
}
跑起来：go run main.go。
这里有两个关键约定：jobs 由生产者关闭（谁发谁关），results 由 WaitGroup 的等待方关闭。channel 是方向性资源，关错地方就是 panic。
注意：jobs 和 results 都用了带缓冲的 channel。如果两个都是无缓冲，主 goroutine 在 jobs <- i 时就会阻塞，而 worker 想往 results 写又没人读，直接死锁。缓冲大小至少要等于任务数，或者像下面那样把发送放进独立 goroutine。
第三步：Fan-out —— 多个消费者抢同一个 channel
上面 for w := 1; w <= numWorkers; w++ 这个循环就是 Fan-out。想动态控制并发数，用带缓冲 channel 当信号量更灵活：
sem := make(chan struct{}, 10) // 最多 10 个并发
for _, task := range tasks {
sem <- struct{}{} // 占坑
go func(t Task) {
defer func() { <-sem }() // 释放
process(t)
}(task)
}
for i := 0; i < cap(sem); i++ {
sem <- struct{}{} // 等所有 goroutine 跑完
}
注意：Go 1.22 之前，for _, task := range 里的循环变量是被所有 goroutine 共享的，必须像上面那样通过参数 t Task 传进去。Go 1.22 起每个迭代独立变量，但显式传参永远是安全写法，建议保留。这个坑用 go vet 也不一定报，靠 go run -race 能抓出来。
第四步：Fan-in —— 多个 channel 合成一个
典型场景：多个数据源并发拉取，结果要统一处理。
func fanIn(ctx context.Context, chs ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
for _, ch := range chs {
wg.Add(1)
go func(c <-chan int) {
defer wg.Done()
for {
select {
case v, ok := <-c:
if !ok {
return
}
select {
case out <- v:
case <-ctx.Done():
return
}
case <-ctx.Done():
return
}
}
}(ch)
}
go func() {
wg.Wait()
close(out) // 唯一关闭 out 的地方
}()
return out
}
调用方统一 for v := range fanIn(ctx, a, b, c) 即可，不用关心源有几个。内层再套一层 select 是为了防止 ctx 取消后 out 没人读、转发 goroutine 卡死在发送上——这就是最容易被忽略的 goroutine 泄漏点。
注意：Fan-in 里 out 只能在一处 close。如果每个转发 goroutine 都 defer close(out)，第一个跑完就 panic: send on closed channel。正确做法永远是用 WaitGroup 等到全部退出再关。
第五步：组合成真实可用的批量处理器
实际业务里还需要错误收集和取消。加一个 error channel 和一个 context.WithCancel：
func Run(ctx context.Context, items []int, workers int) ([]int, []error) {
jobs := make(chan int)
resCh := make(chan int)
errCh := make(chan error, len(items))
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for it := range jobs {
if it < 0 {
errCh <- fmt.Errorf("bad item: %d", it)
continue
}
select {
case resCh <- it * 2:
case <-ctx.Done():
return
}
}
}()
}
go func() {
defer close(jobs)
for _, it := range items {
select {
case jobs <- it:
case <-ctx.Done():
return
}
}
}()
go func() {
wg.Wait()
close(resCh)
close(errCh)
}()
var out []int
for r := range resCh {
out = append(out, r)
}
var errs []error
for e := range errCh {
errs = append(errs, e)
}
return out, errs
}
关键点：errCh 用带缓冲 len(items)，保证 worker 写错误永不阻塞；jobs 无缓冲但由独立 goroutine 发送，主 goroutine 不会卡住。
验证一下有没有并发问题：
go run -race main.go
go vet ./...
小结
- Worker Pool = 固定消费者 + 任务 channel，核心价值是限流而不是提速。
- Fan-out 是分发动作，用 channel 信号量或固定循环都能实现；Fan-in 是汇总动作，靠 WaitGroup + 单一 close 收口。
- channel 归属要清晰：发送方关任务 channel，等待方关结果 channel，一个 channel 只关一次。
- 想不死锁：任务和结果 channel 都加缓冲，或把发送动作放进独立 goroutine。
- 想不泄漏：所有阻塞的发送和接收都要套
select { case <-ctx.Done(): return }。
- 上生产前必跑
go run -race，循环变量捕获和并发写切片这两类 bug 只有它能抓。
正式会员
认证极客
