go语言中pipeline是一种并发处理模式，通过分阶段处理数据流提高效率。它由生产者、处理器和消费者组成，各阶段用channel连接。构建时要注意：每个stage专注自身逻辑；channel设缓冲避免阻塞；控制goroutine生命周期防泄露。实际案例包括生成整数、平方处理、过滤条件值、最终求和。关闭pipeline时需引入done channel，监听退出信号以释放资源。以上步骤确保pipeline高效稳定运行。
在Go语言中，Pipeline是一种常见的并发处理模式，特别适合分阶段处理数据流的场景。它的核心思想是将一个任务拆分成多个步骤，每个步骤由独立的goroutine负责，并通过channel传递中间结果。
这种方式既能提高程序执行效率，又能保持代码结构清晰，尤其适用于数据转换、批量处理、流水线式计算等任务。
Pipeline的基本结构
典型的Pipeline由三个部分组成：生产者（source）、处理器（stage）和消费者（sink）。各阶段之间通过channel连接，前一阶段的输出作为后一阶段的输入。
立即学习“go语言免费学习笔记（深入）”；
比如，你有一个整数切片，想先对每个数字做平方，再过滤掉小于100的结果，最后求和。这些操作就可以分别放在不同的stage中完成。
构建时要注意以下几点：
- 每个stage只关注自己的处理逻辑
- channel要设置合适的缓冲大小，避免阻塞
- 需要控制goroutine生命周期，防止泄露
分阶段使用Channel处理的实际案例
以一个实际例子来看，假设我们要实现这样一个流程：
- 生成一组随机整数
- 将这些整数平方
- 过滤出大于100的值
- 最后将所有符合条件的值累加并输出
我们可以为每个步骤创建一个函数，每个函数返回一个channel用于接收数据。
func gen(nums ...int) <-chan int {
out := make(chan int)
go func() {
for _, n := range nums {
out <- n
}
close(out)
}()
return out
}
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
out <- n * n
}
close(out)
}()
return out
}
func filter(in <-chan int, fn func(int) bool) <-chan int {
out := make(chan int)
go func() {
for n := range in {
if fn(n) {
out <- n
}
}
close(out)
}()
return out
}然后把这些stage串联起来：
c := gen(10, 15, 3, 8, 12)
squared := square(c)
filtered := filter(squared, func(n int) bool {
return n > 100
})
sum := 0
for n := range filtered {
sum += n
}
fmt.Println(sum)这个例子虽然简单，但完整地展示了如何用channel串联起各个处理阶段，形成一条清晰的数据流水线。
如何优雅关闭Pipeline
在实际使用中，Pipeline可能需要提前退出，比如用户中断、错误发生等情况。这时如果不妥善关闭，容易造成goroutine泄漏或channel阻塞。
一个常用的做法是引入一个done channel，在各个stage中监听它，一旦收到信号就立即退出。
例如改进上面的square函数：
func square(done <-chan struct{}, in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
select {
case out <- n * n:
case <-done:
return
}
}
}()
return out
}主流程中创建一个done channel，并在最后关闭它：
done := make(chan struct{})
defer close(done)
c := gen(done, 10, 15, 3, 8, 12)
squared := square(done, c)
filtered := filter(done, squared, func(n int) bool {
return n > 100
})
sum := 0
for n := range filtered {
sum += n
}
fmt.Println(sum)这样可以确保即使中途退出，所有goroutine都能及时释放资源，不会留下隐患。
总结
构建Golang中的Pipeline，关键在于合理划分处理阶段，利用channel连接各个步骤，并注意控制goroutine的生命周期。对于数据流清晰、可分阶段的任务来说，这种模式非常实用且高效。
基本上就这些了。只要理解了分阶段处理的思路和channel的使用方式，就能灵活应用到各种并发任务中。
