2、并发中间件设计模式：Fan-in、Fan-out、Pipeline、Worker Pool 模式详解

在构建高性能 Go 并发中间件时，选择正确的并发模式直接决定了系统的吞吐量、资源利用率和可维护性。本章深入剖析四种核心设计模式：Fan-out（扇出）、Fan-in（扇入）、Pipeline（流水线）和 Worker Pool（工作池）。这些模式是 Go 并发编程的基石，也是后续 pprof 内存分析优化的重要场景。

2.1 模式总览与对比

四种模式分别解决不同的并发问题，下表从核心思想、适用场景和内存特征三个维度进行对比：

2.2 Fan-out 模式：并行任务分发

Fan-out 模式将一个输入 channel 的数据分发到多个 goroutine 中并行处理。每个 goroutine 从同一个 channel 读取数据，Go 的 channel 机制天然支持多消费者竞争读取。

// Fan-out 模式示例：将输入任务分发到 4 个 worker
func fanOut(input <-chan Task, numWorkers int) []<-chan Result {
    outputs := make([]<-chan Result, numWorkers)
    for i := 0; i < numWorkers; i++ {
        outputs[i] = worker(input)
    }
    return outputs
}

func worker(tasks <-chan Task) <-chan Result {
    results := make(chan Result, 10) // 带缓冲 channel
    go func() {
        defer close(results)
        for task := range tasks {
            results <- process(task)
        }
    }()
    return results
}

2.3 Fan-in 模式：多路结果聚合

Fan-in 模式将多个输入 channel 合并为一个输出 channel，通常使用 sync.WaitGroup 或 reflect.Select 实现。它是 Fan-out 的自然对称模式。

// Fan-in 模式示例：合并多个结果 channel
func fanIn(channels ...<-chan Result) <-chan Result {
    out := make(chan Result)
    var wg sync.WaitGroup
    wg.Add(len(channels))

    for _, ch := range channels {
        go func(c <-chan Result) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(ch)
    }

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

2.4 Pipeline 模式：阶段化处理链

Pipeline 模式将处理流程拆分为多个阶段（stage），每个阶段由独立的 goroutine 执行，通过 channel 传递数据。每个阶段只关注自己的输入和输出，形成清晰的职责边界。

// Pipeline 模式示例：三阶段流水线
func pipeline(input <-chan int) <-chan int {
    // 阶段1：乘以 2
    stage1 := func(in <-chan int) <-chan int {
        out := make(chan int)
        go func() {
            defer close(out)
            for v := range in {
                out <- v * 2
            }
        }()
        return out
    }

    // 阶段2：加 1
    stage2 := func(in <-chan int) <-chan int {
        out := make(chan int)
        go func() {
            defer close(out)
            for v := range in {
                out <- v + 1
            }
        }()
        return out
    }

    // 阶段3：过滤偶数
    stage3 := func(in <-chan int) <-chan int {
        out := make(chan int)
        go func() {
            defer close(out)
            for v := range in {
                if v%2 == 0 {
                    out <- v
                }
            }
        }()
        return out
    }

    return stage3(stage2(stage1(input)))
}

2.5 Worker Pool 模式：可控并发度

Worker Pool 模式维护固定数量的 worker goroutine，从任务队列中获取任务并执行。这是生产环境中最常用的并发模式，因为它能精确控制并发度，避免 goroutine 无限制创建。

// Worker Pool 模式示例
type Pool struct {
    tasks    chan Task
    results  chan Result
    wg       sync.WaitGroup
    quit     chan struct{}
}

func NewPool(numWorkers int, bufferSize int) *Pool {
    return &Pool{
        tasks:   make(chan Task, bufferSize),
        results: make(chan Result, bufferSize),
        quit:    make(chan struct{}),
    }
}

func (p *Pool) Start(numWorkers int) {
    for i := 0; i < numWorkers; i++ {
        p.wg.Add(1)
        go p.worker()
    }
}

func (p *Pool) worker() {
    defer p.wg.Done()
    for {
        select {
        case task, ok := <-p.tasks:
            if !ok {
                return
            }
            p.results <- task.Execute()
        case <-p.quit:
            return
        }
    }
}

func (p *Pool) Submit(task Task) {
    p.tasks <- task
}

func (p *Pool) Stop() {
    close(p.quit)
    p.wg.Wait()
    close(p.results)
}

2.6 模式组合与架构图

在实际的并发中间件中，这四种模式通常组合使用。下图展示了一个典型的 Fan-out → Pipeline → Fan-in 组合架构：

2.7 模式选择决策流程

在实际项目中，如何选择最合适的并发模式？以下决策流程可以帮助你快速判断：

1. 是否需要控制并发度？ → 是 → Worker Pool；否 → 进入下一步
2. 处理流程是否可以拆分为多个阶段？ → 是 → Pipeline；否 → 进入下一步
3. 是否需要并行处理同一数据源？ → 是 → Fan-out；否 → 进入下一步
4. 是否需要合并多个数据源？ → 是 → Fan-in；否 → 考虑单 goroutine 处理

2.8 内存分析关注点总结

使用 pprof 对这四种模式进行内存分析时，重点关注以下指标：