Go语言怎么实现goroutine之间的扇入扇出模式-Golang-PHP中文网

扇出是“一个输入，多个 worker 并发处理”，需用分发 goroutine 将数据复制到多个 channel 以避免竞争；避免泄漏须用 sync.waitgroup 等待完成、带缓冲 channel 控制并发、worker 捕获 panic。

go语言怎么实现goroutine之间的扇入扇出模式

扇出就是从一个 goroutine 启动多个 worker goroutine 并发处理任务。常见错误是没控制并发数、没等所有 worker 结束就退出，或者 worker panic 导致主流程卡死。

关键点：用 sync.WaitGroup 等待所有 worker 完成；用带缓冲的 channel 控制并发上限；worker 必须处理 panic，否则会静默终止，WaitGroup.Done() 不会被调用。

扇入是把多个 goroutine 的输出 channel 合并到一个 channel 上。最简方式是起一个 goroutine 逐个读取，但容易因某个 channel 永不关闭而卡住；正确做法是用 select 配合 done channel 实现可取消的多路复用。

典型错误：写成 for v := range ch1 { out + <code>for v := range ch2 { out ，这会导致第二个循环永远等不到关闭。

Go语言（Golang）1.26.0

Go语言（Golang）1.26.0版本官方下载，版本号 1.26.0，适合旧项目维护、兼容性测试和指定版本开发环境搭建。

下载

每个输入 channel 单独起 goroutine 读取，并在读完后 close(out)？不行 —— 多个 goroutine close 同一个 channel 会 panic
正确模式：用 sync.WaitGroup 计数输入源数量，所有源读完后 close 输出 channel
务必加 done channel 到 select 分支，否则上游提前退出时，扇入 goroutine 可能永远阻塞在某个

示例核心结构：

go func() {
    defer wg.Done()
    for {
        select {
        case v, ok :=

原生扇入扇出不保序，也不传错。如果你需要“按输入顺序返回结果”或“任一 worker 出错就中止全部”，就得额外设计信号通道和上下文控制。

保序难点在于：worker 执行时间不同，先启动的不一定先返回。错误传播则依赖 context.Context 的取消能力，而不是靠 channel 关闭来判断失败。

要保序：给每个任务加 id int，worker 返回 struct{ ID int; Result R; Err error }，扇入端用 map 缓存，用最小堆或排序切片按 ID 合并
要传播错误：所有 worker 监听同一个 ctx.Done()，并在发现 ctx.Err() != nil 时立即 return；扇入端收到第一个非 nil error 就 cancel context
别用 recover() 捕获 panic 后塞进结果 channel —— 这会让调用方难以区分是业务错误还是崩溃，应让 panic 触发进程级日志并由上层监控捕获
worker 内部不要自己重试；重试逻辑应放在扇出前的任务分发层，否则多个 worker 可能重复处理同一任务

多数人能写出可运行的扇入扇出代码，但线上一压就出问题，往往栽在这三点上：

输入 channel 没做限流，上游生产太快导致内存暴涨 —— 用 semaphore 或带缓冲 channel 控制未消费任务数，缓冲长度 ≤ 并发 worker 数 × 期望平均处理耗时 / 平均生产间隔
worker panic 后没有记录日志，只靠 recover() 吞掉 —— 致命：panic 信息丢失，无法定位哪类任务触发崩溃；应至少打 log.Printf("worker panic: %v", err)
扇入端用 for range out 读结果，但没设超时或取消 —— 如果某个 worker 卡死或网络 hang 住，整个消费流程就停了；必须配合 context.WithTimeout 使用

扇入扇出不是模式套用，而是对 channel 生命周期、goroutine 生存期、错误边界的精细控制。写完记得跑压测，重点看 goroutine 数是否随任务量线性增长、内存是否持续上升、cancel 是否真正生效。