Go 并发流水线：用 Fan-in/Fan-out 把流式数据处理做稳

什么是 Fan-out 和 Fan-in

Fan-out 指一个数据源被多个 worker 同时消费，用来并行处理；Fan-in 指把多个 channel 的结果合并回一个 channel，方便下游统一读取。两者拼起来就是一条并发流水线：生产、并行加工、汇总。

Fan-out：一个源分发给多个 worker

下面的 gen 产生一批整数，三个 worker 各自从同一个 channel 取任务，天然负载均衡。

package main

import "fmt"

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

func main() {
    for r := range square(gen(2, 3, 4)) {
        fmt.Println(r)
    }
}

Fan-in：把多个 channel 合并

把 N 个相同类型的 channel 合并成一个，下游只面对一个出口。

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

完整流水线：多 worker 并行加工再汇总

启动 3 个 square worker，各自处理 gen 产出的数据，最后用 merge 收口。数据量越大，并行收益越明显。

func main() {
    in := gen(1, 2, 3, 4, 5, 6, 7, 8)

    c1 := square(in)
    c2 := square(in)
    c3 := square(in)

    for r := range merge(c1, c2, c3) {
        fmt.Println(r)
    }
}

加 context 做取消和超时

真实场景里要能随时停下整条流水线。把 context 透传进去，下游在 select 里监听 done，收到取消就退出，避免 goroutine 泄漏。

func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for {
            select {
            case <-ctx.Done():
                return
            case n, ok := <-in:
                if !ok {
                    return
                }
                out <- n * n
            }
        }
    }()
    return out
}

错误处理要点

流水线里推荐用一个 error channel 单独传错误，不要在数据 channel 里塞错误值。主 goroutine 用一个 errgroup 或独立 select 收集错误，一旦致命错误就 cancel 整个 context，让所有 worker 有序退出。这样数据通道保持干净，取消逻辑也集中可控。