如何在Golang中实现fan-in和fan-out模式_合并和分发多个channel数据

fan-out 是将一个任务分发给多个 goroutine 并行处理,fan-in 是合并多个 channel 输出为一个;二者组合构建高吞吐数据流水线,需用 WaitGroup 或 context 防泄漏。

在 Go 中,fan-out(扇出)和 fan-in(扇入)是处理并发数据流的经典模式:fan-out 指将一个任务或数据源分发给多个 goroutine 并行处理;fan-in 指将多个 channel 的输出合并为一个 channel,供下游统一消费。二者常组合使用,构建高吞吐、可扩展的数据流水线。

fan-out:将输入分发到多个 goroutine

核心是启动多个 worker goroutine,从同一个输入 channel 读取数据,并行处理。需注意防止 goroutine 泄漏和竞态,通常配合 sync.WaitGroupcontext 控制生命周期。

示例:把一个整数切片分发给 3 个 worker,各自计算平方并发送到各自的输出 channel:

func fanOut(in <-chan int, workers int) []<-chan int {
    out := make([]<-chan int, workers)
    for i := 0; i < workers; i++ {
        ch := make(chan int, 10)
        out[i] = ch
        go func(ch chan<- int) {
            for v := range in {
                ch <- v * v // 并行处理
            }
            close(ch)
        }(ch)
    }
    return out
}

⚠️ 注意:上面写法中,in 被所有 goroutine 共享读取,Go 的 channel 是并发安全的,多个 goroutine 可以同时从同一 读 —— 这正是 fan-out 的基础。但要确保上游不会无限发送,否则 worker 会永远阻塞在 range in

fan-in:合并多个 channel 到单个输出

fan-in 的关键是“非阻塞地监听多个 channel”,标准做法是用 select + for 循环轮询所有输入 channel。为避免某个 channel 关闭后持续空转,需动态管理活跃 channel 列表(或用更简洁的 reflect.Select,但一般不推荐)。

推荐方式:每个输入 channel 启动一个 goroutine 转发到统一的 fan-in channel,并在转发结束后关闭该 goroutine:

func fanIn(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    wg.Add(len(cs))
for _, c := range cs {
    go func(c <-chan int) {
        defer wg.Done()
        for v := range c {
            out <- v
        }
    }(c)
}

// 所有输入 channel 关闭后,关闭输出 channel
go func() {
    wg.Wait()
    close(out)
}()
return out

}

✅ 优点:简洁、无 busy-wait、自动关闭输出 channel;缺点:无法保证输出顺序(因 goroutine 调度不确定)。

组合 fan-out + fan-in 构建完整流水线

典型场景:读取一批 URL,fan-out 给多个 HTTP client 并发请求,再 fan-in 汇总响应结果。

  • 输入 channel 发送 URL 字符串
  • fan-out 启动 N 个 fetcher goroutine,每个调用 http.Get
  • 每个 fetcher 将结果(如 status code 或 error)发到自己的 output channel
  • fan-in 合并所有 output channel,返回统一结果流

关键点:每个 worker 应独立处理错误,不因单个失败中断整个流程;建议用结构体封装结果(含 data/error),避免 channel 类型混杂。

进阶技巧与注意事项

  • context.Context 统一取消:传入 context 到每个 worker,在 select 中监听 ctx.Done(),实现超时或主动中断
  • 限流控制:fan-out 前加 buffer channel 或使用 semaphore(如带缓冲的 channel 模拟信号量),防瞬间启动过多 goroutine
  • 避免内存泄漏:确保所有 goroutine 都有退出路径,尤其注意 channel 未关闭导致 range 永不结束
  • 替代方案:对简单合并,可用 golang.org/x/exp/slices(Go 1.21+)辅助,但 channel 流水线仍是原生、零依赖的最佳实践

不复杂但容易忽略:fan-in 的输出 channel 必须由某个 goroutine 显式关闭,否则下游 range 会永久阻塞;fan-out 的 worker 数量不是越多越好,应结合 CPU 核心数与 I/O 特性权衡。