管道模式利用goroutine和channel实现数据流水处理，由源头、中间处理器和汇点构成，各阶段通过channel传递数据，需正确关闭channel以避免泄漏。
在Go语言中，并发编程通过goroutine和channel的组合变得简洁高效。管道（pipeline）模式正是利用这一特性，将数据流经多个处理阶段，每个阶段由一个或多个goroutine完成，阶段之间通过channel传递数据。这种模式适用于数据处理流水线、ETL任务、消息过滤等场景。
理解管道的基本结构
一个典型的管道由三部分组成：源头（source）、中间处理器（stages）和汇点（sink）。数据从源头产生，依次经过多个处理阶段，最终被消费。
关键点在于每个阶段都从输入channel读取数据，处理后写入输出channel。为了防止goroutine泄漏，需确保所有channel都被正确关闭，并且接收方能感知到结束信号。
- 使用
close()显式关闭不再发送数据的channel - 接收方可通过
value, ok := 判断channel是否已关闭 - 多路复用时配合
sync.WaitGroup等待所有goroutine完成
构建带缓冲的多阶段处理管道
实际应用中，某些阶段可能处理较慢，导致阻塞。为提升吞吐量，可对channel设置缓冲，或将独立任务并行化。
立即学习“go语言免费学习笔记（深入）”；
以下是一个将整数平方后筛选偶数的例子：
func gen(nums ...int) out := make(chan int, len(nums))go func() {
for _, n := range nums {
out }
close(out)
}()
return out
}
func square(in
out := make(chan int)
go func() {
defer close(out)
for n := range in {
out
}
}()
return out
}
func filterEven(in
out := make(chan int)
go func() {
defer close(out)
for n := range in {
if n%2 == 0 {
out
}
}
}()
return out
}
调用方式：result := filterEven(square(gen(1,2,3,4)))，数据像水流一样穿过各个阶段。
处理错误与资源清理
真实系统中，某个阶段可能出错，需要中断整个流程并释放资源。此时可引入context包来统一控制生命周期。
将context.Context作为参数传入每个阶段，当发生错误或超时时，取消context，所有监听它的goroutine应尽快退出。
- 每个阶段检查
ctx.Done()以响应取消信号 - 使用
select同时监听数据channel和context的done channel - 错误信息可通过单独的error channel返回给主协程
这样能保证程序在异常情况下也能优雅退出，避免goroutine堆积。
合并多个数据源（fan-in）与分发任务（fan-out）
为了提高性能，可以在瓶颈阶段启用多个worker并发处理，然后将结果合并。这就是fan-out/fan-in模式。
例如，在square阶段启动多个goroutine处理输入数据，再通过一个channel汇总结果：
func merge(cs [] var wg sync.WaitGroupout := make(chan int)
output := func(c
for n := range c {
out
}
wg.Done()
}
wg.Add(len(cs))
for _, c := range cs {
go output(c)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
结合makeWorkers()函数生成多个处理goroutine，可显著提升计算密集型任务的效率。
基本上就这些。Go的管道模式强大而灵活，核心是合理设计数据流向与生命周期管理。只要把握好channel的关闭时机和错误传播机制，就能写出清晰可靠的并发代码。
