Go管道模式[编辑 | 编辑源代码]

Go管道模式（Pipeline Pattern）是Go语言中利用通道（Channel）实现的一种函数式编程范式，用于将多个处理阶段串联起来，形成数据流处理链。该模式借鉴了Unix的管道思想（如`cmd1 | cmd2 | cmd3`），通过并发和模块化设计提高代码的可读性和性能。

核心概念[编辑 | 编辑源代码]

管道模式由以下要素构成：

阶段（Stage）：每个处理单元（函数）负责特定任务，通过通道连接。
通道（Channel）：阶段间的数据传输媒介，支持同步或异步通信。
扇出/扇入（Fan-out/Fan-in）：多个协程处理同一阶段（扇出），结果合并到单一通道（扇入）。

数学表达[编辑 | 编辑源代码]

管道可形式化为函数组合： $P = f_{n} \circ \dots \circ f_{2} \circ f_{1}$ 其中 $f_{i}$ 代表各阶段函数， $\circ$ 表示通过通道连接。

基础示例[编辑 | 编辑源代码]

以下示例展示三阶段管道：生成数字→平方→过滤偶数。

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 sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

// 过滤偶数
func even(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            if n%2 == 0 {
                out <- n
            }
        }
        close(out)
    }()
    return out
}

func main() {
    // 构建管道：gen → sq → even
    pipeline := even(sq(gen(1, 2, 3, 4, 5)))
    
    // 消费结果
    for result := range pipeline {
        fmt.Println(result) // 输出：4 16
    }
}

输出说明：

输入`1,2,3,4,5`经平方后变为`1,4,9,16,25`
过滤后仅保留`4,16`

高级模式[编辑 | 编辑源代码]

扇出/扇入[编辑 | 编辑源代码]

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

func main() {
    in := gen(1, 2, 3, 4)
    
    // 启动两个sq worker
    c1 := sq(in)
    c2 := sq(in)
    
    // 合并结果
    for n := range merge(c1, c2) {
        fmt.Println(n) // 输出：1 4 9 16（顺序可能不同）
    }
}

错误处理[编辑 | 编辑源代码]

通过额外通道传递错误：

type Result struct {
    Value int
    Err   error
}

func safeSq(in <-chan int) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for n := range in {
            if n > 100 { // 模拟错误条件
                out <- Result{Err: fmt.Errorf("value too large: %d", n)}
                continue
            }
            out <- Result{Value: n * n}
        }
    }()
    return out
}

应用场景[编辑 | 编辑源代码]

日志处理：解析→过滤→聚合日志流
ETL管道：数据抽取→转换→加载
实时计算：传感器数据→清洗→分析

性能考量[编辑 | 编辑源代码]

通过缓冲通道平衡生产/消费速度：make(chan int, 100)
使用context.Context实现超时控制
避免通道泄漏：确保所有协程能正常退出

最佳实践[编辑 | 编辑源代码]

1. 明确阶段职责，保持单一功能 2. 使用defer close(channel)防止死锁 3. 监控协程数量防止资源耗尽

模板:Note