如何使用Golang pipeline模式处理数据流_分阶段处理协程任务
Go 语言 pipeline 模式通过 channel 串联“生产-处理-消费”阶段，各阶段为独立函数，接收输入 channel 并返回输出 channel，内部启 goroutine 处理，输入类型统一且关闭后下游自然退出。
Go 语言中的 pipeline 模式是一种通过组合多个阶段（每个阶段由一个或多个 goroutine 构成）来处理数据流的惯用方式，核心是使用 channel 串联“生产-处理-消费”流程，实现解耦、并发与可控背压。
定义清晰的阶段函数，每个阶段只做一件事
每个 pipeline 阶段应是一个独立函数，接收输入 channel，返回输出 channel，内部启动 goroutine 处理数据。避免在单个函数中混入多种逻辑。
- 输入 channel 类型统一（如
chan int），关闭后下游可自然退出 - 用
for range读取输入，显式关闭输出 channel（除非需持续接收） - 示例：过滤偶数阶段
out := make(chan int)
go func() {
defer close(out)
for v := range in {
if v%2 == 0 {
out
}
}
}()
return out
}
用 channel 连接阶段，形成线性或分叉数据流
将前一阶段的输出 channel 直接作为下一阶段的输入，构成流水线。支持串行（A→B→C）或扇出/扇入（如多个 worker 并行处理同一输入）。
- 串行写法简洁：
result := stageC(stageB(stageA(source))) - 扇出：用 for 启动多个相同 stage 的 goroutine，共用一个输入 channel
- 扇入：用
merge函数合并多个 output channel 到一个 channel（可用
select+goroutine实现）
处理错误和终止：用 done channel 控制生命周期
原始 pipeline 在某个阶段 panic 或阻塞时可能造成 goroutine 泄漏。引入 done 参数可安全取消所有阶段。
- 每个阶段在
for range中加入select检查done - 上游提前关闭或发送 cancel 信号，下游能及时退出
- 推荐用
context.Context替代裸donechannel，便于超时与层级取消
注意缓冲与背压：避免 channel 堆积或死锁
无缓冲 channel 要求收发双方同时就绪，易导致阻塞；过度缓冲又可能吃光内存。合理选择缓冲策略：
- I/O 密集型阶段（如 HTTP 请求）建议带缓冲（如
make(chan T, 16)）缓解瞬时延迟 - CPU 密集型或需强顺序场景用无缓冲，天然实现同步与限流
- 关键：确保每个阶段都消费完输入，否则上游会因无法发送而挂起
