Golang并发管道：Fan-in Fan-out实现详解

本文深入剖析了Go语言中并发管道模式的核心实践——Fan-in与Fan-out的正确实现，直击常见误区：直接让多个goroutine共享读取同一channel会导致数据丢失和竞态，而盲目合并channel则易引发goroutine泄漏或永久阻塞；文章强调安全Fan-out必须为每个goroutine显式复制数据流并确保原始channel及时关闭，Fan-in则需结合goroutine、WaitGroup与context精确控制生命周期，从而构建高可靠、可终止、无资源泄漏的并发数据处理流水线。

Fan-out：怎么安全地把一个channel分发给多个goroutine

直接用同一个 chan int 同时塞给 3 个 goroutine 读，不加控制会丢数据——因为没有同步机制，谁抢到谁读，但没人保证每个值都被所有 goroutine 消费一次。

正确做法是显式复制数据流：启动多个 goroutine，每个都从原始 channel 里读，各自处理。关键点在于「原始 channel 必须被关闭」，否则接收方可能永远阻塞。

• 原始 channel 类型要是 <-chan T（只读），避免下游误写
• 每个 fan-out goroutine 应该用 for range 遍历，自动响应 close
• 别在 fan-out 里再起 goroutine 去转发，容易漏关、死锁

func fanOut(in <-chan int, n int) []<-chan int {
    out := make([]<-chan int, n)
    for i := 0; i < n; i++ {
        out[i] = worker(in)
    }
    return out
}
<p>func worker(in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for v := range in {
out <- v * 2 // 示例处理
}
}()
return out
}</p>

Fan-in：多个channel怎么合并成一个，还不卡死

常见错误是用 select 轮询多个 channel，但没加 default 或超时，一旦某个 channel 永久阻塞（比如没关），整个 fan-in 就停住。

更稳的方式是为每个输入 channel 启一个 goroutine，统一往一个输出 channel 写；fan-in 的 channel 必须是带缓冲的，或确保有接收方在跑，否则写入会阻塞 goroutine。

• 输入 channel 全部用 <-chan T，防止意外写入
• fan-in goroutine 必须在所有输入 channel 关闭后才 close 输出 channel，可用 sync.WaitGroup 计数
• 如果某输入 channel 可能永远不关（比如日志流），得加 context 控制生命周期

func fanIn(ctx context.Context, cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    wg.Add(len(cs))
<pre class="brush:php;toolbar:false;">for _, c := range cs {
    go func(c &lt;-chan int) {
        defer wg.Done()
        for v := range c {
            select {
            case out &lt;- v:
            case &lt;-ctx.Done():
                return
            }
        }
    }(c)
}

go func() {
    wg.Wait()
    close(out)
}()
return out

}

为什么用 context.Context 而不是靠 channel 关闭来退出

channel 关闭只能表示“数据源结束”，但没法表达“我不要了”——比如用户按 Ctrl+C、HTTP 请求超时、上游服务已下线。这时候靠关 channel 不顶用，goroutine 还在等下一个值。

context.WithTimeout 或 context.WithCancel 是唯一能跨 goroutine 主动中断的机制。所有 fan-out/fan-in 的 goroutine 都得监听 ctx.Done()，并在收到信号后立即清理、退出。

• 别在 goroutine 里忽略 <-ctx.Done()，尤其在循环中
• fan-in 合并时，select 必须把 <-ctx.Done() 放在第一分支，避免被其他 channel 抢占
• 用 context.WithCancel 时，cancel 函数要传出去，由调用方决定何时触发

实际管道链中容易漏掉的资源泄漏点

最常被忽略的是：中间 channel 没人读，但上游还在拼命写。哪怕用了 context，如果 fan-out 的 goroutine 早退了，而 fan-in 还没启动，数据就堆在 channel 缓冲区里，goroutine 和内存全卡住。

• 所有 channel 缓冲大小必须明确设值，make(chan int, 0) 和 make(chan int, 1) 行为完全不同
• 如果管道某段失败（比如解析 JSON 出错），别只关自己的输出 channel，得显式 cancel context，让整条链退出
• 测试时用 runtime.NumGoroutine() 看 goroutine 数是否随请求增长，涨了就是泄漏

管道不是搭积木，是拧螺丝——每一段的关闭时机、context 传递、错误传播，都得对齐，差一点，整条链就悬在半空。

本篇关于《Golang并发管道：Fan-in Fan-out实现详解》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于Golang的相关知识，请关注golang学习网公众号！