Go面试必问：并发模式 Worker Pool vs Pipeline vs FanOut/FanIn 对比

Go面试必问并发模式 Worker Pool vs Pipeline vs FanOut/FanIn 对比导语Go 的并发编程有三把斧Worker Pool、Pipeline、FanOut/FanIn。面试中经常出现请对比这三种模式的应用场景和性能特点这类问题。很多候选人对三种模式的理解停留在都用 channel 通信却说不清各自的适用场景、背压处理能力、以及错误传播机制的差异。本文用生产级代码示例系统对比这三种模式的底层原理与实践要点。核心技术知识点讲解1. 三种模式的核心定义模式核心思想典型拓扑Worker Pool固定数量的 worker 从共享任务队列取任务1个任务队列 → N个worker → 1个结果收集器Pipeline数据流式处理每个 stage 是纯函数生产者 → stage1 → stage2 → … → 消费者FanOut/FanIn多个 worker 并行处理结果汇聚到单个 channel1个输入 → N个workerFanOut→ 1个汇聚FanIn2. Worker Pool 的底层原理Worker Pool 本质是任务队列 固定数量 GoroutinetypeWorkerPoolstruct{taskschanfunc()// 任务队列wg sync.WaitGroup workersint}关键设计点任务队列用buffered channel容量决定排队长度worker 数量固定避免 Goroutine 暴涨通过context.Context实现优雅关闭3. Pipeline 的底层原理Pipeline 每个 stage 是纯函数 channel 串联// 典型 stage 签名funcstage(in-chanT)-chanR关键设计点每个 stage 负责启动自己的 Goroutine 并在退出时 close(outputChan)下游 stage 通过range channel自动检测关闭需要context 传播取消避免 stage 泄漏4. FanOut / FanIn 的底层原理FanOut多个 Goroutine 从同一个 input channel读取自动负载均衡Go channel 保证公平调度FanIn多个 Goroutine 向同一个 output channel写入或用select汇聚多个 channel// FanIn用 select 汇聚多个 channelfuncfanIn(chs...-chanT)-chanT{out:make(chanT)varwg sync.WaitGroup wg.Add(len(chs))for_,ch:rangechs{gofunc(c-chanT){deferwg.Done()forv:rangec{out-v}}(ch)}gofunc(){wg.Wait()close(out)}()returnout}5. 三种模式的核心区别对照维度Worker PoolPipelineFanOut/FanIn任务类型独立任务无顺序要求流式数据有顺序独立任务需汇聚结果背压处理通过 buffered channel 容量每个 stage 的 buffer 决定FanIn 的 output buffer 决定错误传播需单独 error channel每个 stage 需传递 error每个 worker 需返回 errorGoroutine 数固定N个worker每个 stage 1个 Goroutine输入1 worker N 汇聚1适用场景后台任务、异步处理ETL、流式计算并行计算、请求扇出实战代码演示/项目案例总结案例1完整 Worker Pool 实现packagemainimport(contextfmtsynctime)typeTaskstruct{IDintJobfunc()errorDonechanerror}typeWorkerPoolstruct{taskschanTask wg sync.WaitGroup ctx context.Context cancel context.CancelFunc}funcNewWorkerPool(workerCount,queueSizeint)*WorkerPool{ctx,cancel:context.WithCancel(context.Background())p:WorkerPool{tasks:make(chanTask,queueSize),ctx:ctx,cancel:cancel,}p.wg.Add(workerCount)fori:0;iworkerCount;i{gop.worker(i)}returnp}func(p*WorkerPool)worker(idint){deferp.wg.Done()for{select{case-p.ctx.Done():fmt.Printf(worker %d 退出\n,id)returncasetask,ok:-p.tasks:if!ok{return}err:task.Job()task.Done-err}}}func(p*WorkerPool)Submit(jobfunc()error)-chanerror{done:make(chanerror,1)p.tasks-Task{Job:job,Done:done}returndone}func(p*WorkerPool)Shutdown(){close(p.tasks)p.wg.Wait()}funcmain(){pool:NewWorkerPool(3,10)varwg sync.WaitGroupfori:0;i5;i{i:i wg.Add(1)gofunc(){deferwg.Done()done:pool.Submit(func()error{time.Sleep(time.Millisecond*time.Duration(i*100))fmt.Printf(任务 %d 完成\n,i)returnnil})-done}()}wg.Wait()pool.Shutdown()fmt.Println(所有任务完成)}案例2完整 Pipeline 实现packagemainimport(contextfmtsync)// 生成器funcgenerator(ctx context.Context,nums...int)-chanint{out:make(chanint)gofunc(){deferclose(out)for_,n:rangenums{select{case-ctx.Done():returncaseout-n:}}}()returnout}// stage1: 平方funcsquare(ctx context.Context,in-chanint)-chanint{out:make(chanint)gofunc(){deferclose(out)forn:rangein{select{case-ctx.Done():returncaseout-n*n:}}}()returnout}// stage2: 过滤偶数funcfilterEven(ctx context.Context,in-chanint)-chanint{out:make(chanint)gofunc(){deferclose(out)forn:rangein{ifn%20{select{case-ctx.Done():returncaseout-n:}}}}()returnout}funcmain(){ctx,cancel:context.WithCancel(context.Background())defercancel()gen:generator(ctx,1,2,3,4,5)sq:square(ctx,gen)result:filterEven(ctx,sq)forv:rangeresult{fmt.Println(v)// 4, 16 (2²4, 4²16)}}案例3FanOut/FanIn 完整实现packagemainimport(fmtsynctime)// Worker 函数FanOutfuncworker(idint,jobs-chanint,resultschan-int,wg*sync.WaitGroup){deferwg.Done()forj:rangejobs{time.Sleep(time.Millisecond*100)// 模拟耗时results-j*2fmt.Printf(worker %d 处理任务 %d\n,id,j)}}// FanIn汇聚多个结果 channelfuncfanIn(chs...-chanint)-chanint{out:make(chanint)varwg sync.WaitGroup wg.Add(len(chs))for_,ch:rangechs{gofunc(c-chanint){deferwg.Done()forv:rangec{out-v}}(ch)}gofunc(){wg.Wait()close(out)}()returnout}funcmain(){jobs:make(chanint,100)results:make(chanint,100)// FanOut启动3个 workervarwg sync.WaitGroupforw:1;w3;w{wg.Add(1)goworker(w,jobs,results,wg)}// 发送任务forj:1;j5;j{jobs-j}close(jobs)// 等待所有 worker 完成然后关闭 resultsgofunc(){wg.Wait()close(results)}()// 消费结果forr:rangeresults{fmt.Println(结果:,r)}}案例4三种模式组合使用生产级场景// 场景日志处理系统// 1. 用 Pipeline 做日志解析解析 → 过滤 → 增强// 2. 用 FanOut/FanIn 做并行处理多个 enrich worker// 3. 用 Worker Pool 做持久化固定数量的 DB writer// 伪代码// gen(logSource) → parseStage → filterStage → enrichFanOut(N) → fanIn → sinkPool(M)开发痛点与报错避坑指南坑1Pipeline 中 Goroutine 泄漏原因某个 stage 退出后上游 stage 还在往 channel 写数据导致阻塞。// ❌ 错误没有 context 取消传播funcstage(in-chanint)-chanint{out:make(chanint)gofunc(){deferclose(out)forn:rangein{out-n*2// 如果 out 的消费者提前退出这里会永远阻塞}}()returnout}// ✅ 正确用 context 或 done channel 传播取消funcstage(ctx context.Context,in-chanint)-chanint{out:make(chanint)gofunc(){deferclose(out)for{select{case-ctx.Done():returncasen,ok:-in:if!ok{return}select{case-ctx.Done():returncaseout-n*2:}}}}()returnout}坑2FanIn 中 nil channel 的处理// ❌ 错误直接往 nil channel 发送会永远阻塞varchchanint// nilch-1// 永远阻塞// ✅ 正确FanIn 时确保每个 channel 都非 nil或用 reflect.Select坑3Worker Pool 的 task channel 未关闭导致 worker 泄漏// ❌ 错误不关闭 tasks channelworker 永远阻塞在 -taskspool:NewWorkerPool(5,10)// 提交任务...// 忘记调用 pool.Shutdown()// ✅ 正确显式关闭pool.Shutdown()// 内部close(tasks) wg.Wait()坑4Pipeline 中 close 顺序错误// 规则谁创建 channel谁负责 close// gen 创建 out channelgen 负责 close// stage 创建自己的 out channelstage 负责 close// ❌ 错误外部 closeout:generator(nums)close(out)// 错误generator 内部还会往 out 写// ✅ 正确generator 内部 defer close(out)坑5FanOut 时任务分配不均Go channel 的调度是公平的但任务耗时差异大时仍可能不均。解决方案// 方案1用 sync.Pool 或 work-stealing 算法复杂// 方案2用 task channel 自然负载均衡Go 内置推荐// 方案3用 errgroup 任务队列适合任务数量已知的场景全文总结技术进阶展望本文系统对比了 Worker Pool、Pipeline、FanOut/FanIn 三种并发模式的底层原理、适用场景和核心区别并提供了生产级完整实现代码。核心要点Worker Pool固定 worker 数适合后台异步任务通过 buffered channel 控制排队Pipeline流式处理每个 stage 是纯函数需正确传播 context 取消FanOut/FanIn并行处理 结果汇聚Go channel 天然支持 FanOut 负载均衡Goroutine 泄漏是三种模式共同的痛点必须通过 context 取消或 done channel 解决进阶方向研究sync.ErrGroup与三种模式的结合错误传播了解golang.org/x/sync/semaphore实现动态 Worker Pool研究golang.org/x/sync/errgroup的SetLimit实现并发数控制Go 1.20深入runtime/channel.go理解 channel 的调度公平性FanOut 负载均衡的底层保证参考文献Go 官方博客《Go Concurrency Patterns: Pipelines and cancellation》2014Sameer AjmaniGo 官方博客《Advanced Go Concurrency Patterns》2013Rob Pike视频《Go 语言设计与实现》— draveness.me/golang/docs/part3-runtime/ch05-scheduler/Go 官方包文档golang.org/x/sync/errgroupUber Go Style Guide — Channel 使用规范与 Goroutine 泄漏防范Go 源码runtime/chan.go— channel 发送/接收的调度逻辑CSDN/知乎Go 并发模式对比实战文章