3、Channel通信模式：扇出(Fan-Out)模式、扇入(Fan-In)模式、Pipeline模式

在Go中间件链路开发中，channel不仅是goroutine之间的通信桥梁，更是构建复杂并发流水线的核心原语。本章深入剖析三种最实用的channel组合模式：扇出(Fan-Out)、扇入(Fan-In)和Pipeline。这三种模式是中间件链路中数据分发、结果聚合与阶段化处理的基石。

3.1 模式概览与适用场景

三种模式分别解决不同的并发编排问题，下表从核心语义、典型场景和关键特征三个维度进行对比：

3.2 扇出(Fan-Out)模式：任务分发与并行处理

扇出模式的核心思想是一对多：一个生产者将数据发送到多个消费者goroutine，每个消费者独立处理完整的数据副本或分片。在中间件链路中，扇出常用于将请求分发到多个校验器、转换器或处理器并行执行。

3.2.1 标准扇出实现

// FanOut 将输入数据分发到多个worker goroutine
func FanOut(input <-chan Task, numWorkers int) []<-chan Result {
    outputs := make([]<-chan Result, numWorkers)
    for i := 0; i < numWorkers; i++ {
        out := make(chan Result, 10)
        outputs[i] = out
        go func(workerID int, out chan<- Result) {
            defer close(out)
            for task := range input {
                // 模拟处理耗时
                result := processTask(workerID, task)
                out <- result
            }
        }(i, out)
    }
    return outputs
}

3.2.2 扇出模式SVG架构图

3.3 扇入(Fan-In)模式：结果聚合与统一消费

扇入模式是扇出的逆操作：将多个输入channel的数据合并到一个输出channel中，由单个消费者统一处理。在中间件链路中，扇入常用于收集多个并行校验器的结果、合并多个数据源或聚合分布式计算的中间结果。

3.3.1 标准扇入实现

// FanIn 将多个输入channel合并为一个输出channel
func FanIn(channels ...<-chan Result) <-chan Result {
    out := make(chan Result, 10)
    var wg sync.WaitGroup
    wg.Add(len(channels))

    // 为每个输入channel启动一个转发goroutine
    for _, ch := range channels {
        go func(c <-chan Result) {
            defer wg.Done()
            for val := range c {
                out <- val
            }
        }(ch)
    }

    // 等待所有转发goroutine结束，关闭输出channel
    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

3.3.2 扇入模式SVG架构图

3.4 Pipeline模式：阶段化处理链

Pipeline模式将数据处理过程拆分为多个阶段(Stage)，每个阶段由一个goroutine处理，数据通过channel在阶段间传递。每个阶段只关注自己的职责，通过channel与上下游解耦。这是中间件链路中最核心的模式，几乎所有的请求处理链都可以抽象为Pipeline。

3.4.1 Pipeline模式实现

// Pipeline 定义处理阶段
type Stage func(<-chan Request) <-chan Response

// BuildPipeline 将多个阶段串联成处理链
func BuildPipeline(stages ...Stage) Stage {
    return func(input <-chan Request) <-chan Response {
        current := input
        for _, stage := range stages {
            // 将当前channel转换为Response类型，再作为下一阶段输入
            current = stage(current)
        }
        // 最后一个阶段的输出即为最终结果
        return current.(<-chan Response)
    }
}

// 使用示例
func main() {
    // 定义三个处理阶段
    stage1 := func(in <-chan Request) <-chan Response {
        out := make(chan Response, 10)
        go func() {
            defer close(out)
            for req := range in {
                out <- validate(req)   // 校验阶段
            }
        }()
        return out
    }

    stage2 := func(in <-chan Response) <-chan Response {
        out := make(chan Response, 10)
        go func() {
            defer close(out)
            for resp := range in {
                out <- enrich(resp)    // 丰富阶段
            }
        }()
        return out
    }

    stage3 := func(in <-chan Response) <-chan Response {
        out := make(chan Response, 10)
        go func() {
            defer close(out)
            for resp := range in {
                out <- persist(resp)   // 持久化阶段
            }
        }()
        return out
    }

    // 构建Pipeline
    pipeline := BuildPipeline(stage1, stage2, stage3)

    // 启动
    input := make(chan Request, 10)
    go func() {
        for i := 0; i < 100; i++ {
            input <- Request{ID: i}
        }
        close(input)
    }()

    results := pipeline(input)
    for result := range results {
        fmt.Println(result)
    }
}

3.4.2 Pipeline模式SVG架构图

3.5 三种模式的组合应用

在实际中间件链路开发中，三种模式很少单独使用，而是组合形成复杂的处理拓扑。以下是一个典型的组合场景：

1. 扇出：将请求分发到多个校验器并行校验
2. 扇入：收集所有校验结果，判断是否全部通过
3. Pipeline：通过校验的请求进入处理链，依次经过转换、增强、持久化阶段

3.6 错误聚合：在channel模式中处理错误

channel模式本身不直接处理错误，但可以通过以下策略实现错误聚合：

// 结果封装策略示例
type Result struct {
    Data interface{}
    Err  error
}

func worker(input <-chan Task) <-chan Result {
    out := make(chan Result, 10)
    go func() {
        defer close(out)
        for task := range input {
            data, err := process(task)
            out <- Result{Data: data, Err: err}
        }
    }()
    return out
}