17370845950

Go 实现 fan-out fan-in 模式需并发启动 worker 处理任务（fan-out），通过 channel 收集结果（fan-in）；关键用 WaitGroup 确保所有 worker 完成后关闭结果 channel，或用 context 控制超时与取消。

用 Go 实现 fan-out fan-in 模式，核心是：启动多个 goroutine 并行处理任务（fan-out），再通过 channel 收集所有结果（fan-in）。关键在于控制并发、避免 goroutine 泄漏、正确关闭 channel 和处理错误。

1. 基础 fan-out fan-in 结构

使用一个输入 channel 分发任务，多个 worker 从中读取；每个 worker 处理完后将结果发到统一的输出 channel。主 goroutine 负责启动 worker、发送任务、接收并汇总结果。

注意：输出 channel 通常由主 goroutine 创建，worker 只写入；要用 goroutine + range 或 sync.WaitGroup 确保所有 worker 完成后再关闭输出 channel。

2. 使用 WaitGroup 控制 worker 生命周期

适合任务数量固定、需等待全部完成的场景。WaitGroup 让主 goroutine 知道何时所有 worker 已退出，从而安全关闭结果 channel。

• 启动前调用 wg.Add(n)，每个 worker 执行完 wg.Done()
• 另起一个 goroutine 调用 wg.Wait() 后关闭结果 channel
• 输入 channel 可以直接 close，worker 用 for job := range jobs 自动退出

3. 使用 context 控制超时和取消

当任务可能阻塞或需提前终止时，把 context.Context 传给 worker。worker 在处理中定期检查 ctx.Done()，及时退出。

• 用 context.WithTimeout 或 context.WithCancel 包装原始 context
• worker 内部用 select { case 响应取消
• 主 goroutine 中 select 等待结果或超时，避免无限等待

4. 错误处理与结果聚合

不要只返回值，要同时返回 error。结果 channel 类型建议为 struct{ Value T; Err error }，避免因某个 worker panic 或出错导致整个流程中断。

• 每个 worker 单独 recover panic，转为 error 发送到结果 channel
• 主 goroutine 遍历结果 channel 时检查 Err 字段，决定是否继续收集还是提前返回
• 汇总逻辑（如 sum、merge、first non-nil）放在主 goroutine 中，保持 worker 职责单一

不复杂但容易忽略：channel 容量、worker 数量设置、是否需要缓冲 channel、如何区分成功/失败结果。实际使用时建议封装成可复用函数，接受任务切片、worker 数、处理函数和上下文作为参数。