Go 并发模式对照表：Worker Pool、Pipeline 和 Fan-out/in 的场景选择

Go 并发模式对照表Worker Pool、Pipeline 和 Fan-out/in 的场景选择一、从能并发到会并发Go 并发模式的选择困境2026 年初某支付平台的核心系统重构中技术团队在并发模式选择上产生了分歧方案 A使用 Worker Pool线程池模式方案 B使用 Pipeline流水线模式方案 C使用 Fan-out/Fan-in分片聚合模式最终他们通过基准测试发现在 CPU 密集型场景下Worker Pool 性能最好在 IO 密集型场景下Fan-out/in 更优而在数据流式处理场景Pipeline 最合适。本文将系统对比 Go 的三大并发模式给出场景选择决策树。二、模式一Worker Pool工作者池适用场景典型场景需要处理大量独立任务如图片压缩、视频转码任务执行时间相对均匀需要控制最大并发数避免资源耗尽不适合场景任务执行时间差异巨大导致 worker 负载不均需要实时响应Worker Pool 有排队延迟实现与优化package main import ( context fmt sync time ) // Task 任务接口 type Task interface { Execute(ctx context.Context) error } // WorkerPool 工作者池 type WorkerPool struct { workers int // worker 数量 taskQueue chan Task // 任务队列 wg sync.WaitGroup ctx context.Context cancel context.CancelFunc } // NewWorkerPool 创建 Worker Pool func NewWorkerPool(workers int, queueSize int) *WorkerPool { ctx, cancel : context.WithCancel(context.Background()) return WorkerPool{ workers: workers, taskQueue: make(chan Task, queueSize), ctx: ctx, cancel: cancel, } } // Start 启动 Worker Pool func (p *WorkerPool) Start() { for i : 0; i p.workers; i { p.wg.Add(1) go p.worker(i) } } // worker 工作协程 func (p *WorkerPool) worker(id int) { defer p.wg.Done() for { select { case -p.ctx.Done(): fmt.Printf(Worker %d stopped\n, id) return case task, ok : -p.taskQueue: if !ok { return } // 执行任务 if err : task.Execute(p.ctx); err ! nil { fmt.Printf(Worker %d: task failed: %v\n, id, err) } } } } // Submit 提交任务 func (p *WorkerPool) Submit(task Task) error { select { case p.taskQueue - task: return nil case -time.After(1 * time.Second): return fmt.Errorf(task queue is full) } } // SubmitWithPriority 提交优先任务插队 func (p *WorkerPool) SubmitWithPriority(task Task) { // 实现优先级队列需要用到 container/heap // 这里简化直接提交 p.taskQueue - task } // Stop 停止 Worker Pool func (p *WorkerPool) Stop() { close(p.taskQueue) p.cancel() p.wg.Wait() } // 使用示例 type CompressTask struct { FilePath string Result chan error } func (t *CompressTask) Execute(ctx context.Context) error { // 模拟压缩任务CPU 密集 select { case -ctx.Done(): return ctx.Err() case -time.After(500 * time.Millisecond): // 压缩完成 fmt.Printf(Compressed: %s\n, t.FilePath) return nil } } func main() { // 创建 Worker PoolCPU 核心数 * 2 numWorkers : 8 pool : NewWorkerPool(numWorkers, 1000) pool.Start() // 提交任务 for i : 0; i 100; i { task : CompressTask{ FilePath: fmt.Sprintf(/path/to/file_%d.jpg, i), Result: make(chan error, 1), } pool.Submit(task) } // 等待完成 time.Sleep(5 * time.Second) pool.Stop() }性能优化技巧// 优化 1: 动态调整 Worker 数量 type DynamicWorkerPool struct { workers int maxWorkers int taskQueue chan Task mu sync.Mutex // ... } func (p *DynamicWorkerPool) adjustWorkers() { // 根据队列长度动态调整 queueLen : len(p.taskQueue) p.mu.Lock() defer p.mu.Unlock() if queueLen 100 p.workers p.maxWorkers { // 队列积压增加 worker p.addWorker() } else if queueLen 10 p.workers p.maxWorkers/2 { // 队列空闲减少 worker p.removeWorker() } } // 优化 2: 任务超时控制 type TimeoutTask struct { Task Timeout time.Duration } func (t *TimeoutTask) Execute(ctx context.Context) error { ctx, cancel : context.WithTimeout(ctx, t.Timeout) defer cancel() done : make(chan error, 1) go func() { done - t.Task.Execute(ctx) }() select { case err : -done: return err case -ctx.Done(): return fmt.Errorf(task timeout) } }三、模式二Pipeline流水线适用场景典型场景数据流处理ETL、日志处理需要多阶段处理如解析 → 转换 → 加载每个阶段的处理速度不同不适合场景任务之间无依赖关系需要低延迟响应实现与优化package main import ( context fmt ) // Stage 流水线阶段 type Stage func(ctx context.Context, in -chan interface{}) -chan interface{} // Pipeline 流水线 type Pipeline struct { stages []Stage } // NewPipeline 创建流水线 func NewPipeline() *Pipeline { return Pipeline{} } // AddStage 添加阶段 func (p *Pipeline) AddStage(stage Stage) { p.stages append(p.stages, stage) } // Run 运行流水线 func (p *Pipeline) Run(ctx context.Context, input -chan interface{}) -chan interface{} { var currentChan -chan interface{} input for _, stage : range p.stages { currentChan stage(ctx, currentChan) } return currentChan } // 示例日志处理流水线 func LogProcessingPipeline() *Pipeline { pipeline : NewPipeline() // 阶段 1: 读取日志 pipeline.AddStage(func(ctx context.Context, in -chan interface{}) -chan interface{} { out : make(chan interface{}) go func() { defer close(out) for logLine : range in { // 解析日志 parsed : parseLog(logLine.(string)) select { case out - parsed: case -ctx.Done(): return } } }() return out }) // 阶段 2: 过滤日志 pipeline.AddStage(func(ctx context.Context, in -chan interface{}) -chan interface{} { out : make(chan interface{}) go func() { defer close(out) for logEntry : range in { // 过滤掉 DEBUG 级别的日志 if logEntry.(LogEntry).Level ! DEBUG { select { case out - logEntry: case -ctx.Done(): return } } } }() return out }) // 阶段 3: 写入存储 pipeline.AddStage(func(ctx context.Context, in -chan interface{}) -chan interface{} { out : make(chan interface{}) go func() { defer close(out) for logEntry : range in { // 写入数据库 err : saveToDB(logEntry.(LogEntry)) if err ! nil { fmt.Printf(Save failed: %v\n, err) } select { case out - logEntry: case -ctx.Done(): return } } }() return out }) return pipeline } // 辅助函数 type LogEntry struct { Level string Message string Time string } func parseLog(line string) LogEntry { // 简化实际 should use regex return LogEntry{Level: INFO, Message: line, Time: time.Now().String()} } func saveToDB(entry LogEntry) error { // 简化实际应写入数据库 fmt.Printf(Saved: %s\n, entry.Message) return nil } func main() { ctx, cancel : context.WithCancel(context.Background()) defer cancel() // 创建输入 input : make(chan interface{}) go func() { for i : 0; i 100; i { input - fmt.Sprintf(Log line %d, i) } close(input) }() // 运行流水线 pipeline : LogProcessingPipeline() output : pipeline.Run(ctx, input) // 消费输出 for result : range output { fmt.Printf(Processed: %v\n, result) } }优化技巧// 优化 1: 添加 buffer减少阻塞 func BufferedStage(bufferSize int) Stage { return func(ctx context.Context, in -chan interface{}) -chan interface{} { out : make(chan interface{}, bufferSize) // 带 buffer // ... } } // 优化 2: 错误处理某阶段失败不影响整体 func FaultTolerantStage(stage Stage) Stage { return func(ctx context.Context, in -chan interface{}) -chan interface{} { out : make(chan interface{}) go func() { defer close(out) for item : range in { func() { defer func() { if r : recover(); r ! nil { fmt.Printf(Stage panic: %v\n, r) // 继续处理下一个不中断流水线 } }() // 执行原 stage简化 // 实际应该包装 stage 的逻辑 }() } }() return out } }四、模式三Fan-out/Fan-in分片聚合适用场景典型场景需要并行处理独立任务如并发请求多个 APIIO 密集型任务网络请求、数据库查询需要聚合结果不适合场景任务有依赖关系需要保证结果顺序实现与优化package main import ( context fmt sync time ) // FanOut 分片将一个输入分发到多个 worker func FanOut(ctx context.Context, input -chan interface{}, workers int) []-chan interface{} { outputs : make([]-chan interface{}, workers) for i : 0; i workers; i { out : make(chan interface{}) outputs[i] out go func(workerID int, out chan interface{}) { defer close(out) for item : range input { // 每个 worker 处理一部分任务 // 简化这里每个 worker 处理所有任务实际应该用负载均衡 select { case out - item: case -ctx.Done(): return } } }(i, out) } return outputs } // FanIn 聚合将多个输入合并为一个输出 func FanIn(ctx context.Context, inputs ...-chan interface{}) -chan interface{} { out : make(chan interface{}) var wg sync.WaitGroup // 启动多个 goroutine每个负责一个输入 for _, in : range inputs { wg.Add(1) go func(in -chan interface{}) { defer wg.Done() for item : range in { select { case out - item: case -ctx.Done(): return } } }(in) } // 所有输入都关闭后关闭输出 go func() { wg.Wait() close(out) }() return out } // 完整示例并发请求多个 API func FetchMultipleAPIs(urls []string) []string { ctx, cancel : context.WithCancel(context.Background()) defer cancel() // 创建输入 channel input : make(chan interface{}) go func() { for _, url : range urls { input - url } close(input) }() // Fan-out: 启动 5 个 worker 并发请求 outputs : FanOut(ctx, input, 5) // Fan-in: 聚合结果 resultChan : FanIn(ctx, outputs...) // 收集结果 results : make([]string, 0) for result : range resultChan { results append(results, result.(string)) } return results } // 优化带错误的 Fan-in func FanInWithErrors(ctx context.Context, inputs ...-chan Result) -chan Result { out : make(chan Result) var wg sync.WaitGroup for _, in : range inputs { wg.Add(1) go func(in -chan Result) { defer wg.Done() for result : range in { select { case out - result: case -ctx.Done(): return } } }(in) } go func() { wg.Wait() close(out) }() return out } type Result struct { Value interface{} Error error }性能对比五、总结Go 并发模式选择决策树选择 Worker Pool✅ CPU 密集型任务✅ 需要控制最大并发数✅ 任务独立且数量大❌ 不适合长任务导致 worker 饥饿选择 Pipeline✅ 数据流处理ETL✅ 多阶段处理✅ 每个阶段速度不同❌ 不适合任务无依赖关系选择 Fan-out/in✅ IO 密集型任务✅ 需要并发处理独立任务✅ 需要聚合结果❌ 不适合需要保证顺序组合使用Worker Pool Pipeline每个 stage 使用 Worker PoolFan-out/in Pipeline分片处理后再流水线处理性能调优建议Worker 数量 CPU 核心数 × 2CPU 密集Worker 数量可以更大IO 密集Channel buffer 大小要合理避免内存浪费一定要处理 context 取消避免 goroutine leak记住没有最好的并发模式只有最适合场景的并发模式。下一篇我们将深入探讨 Function Calling 模式选择指南。