Golang任务流水线通过channel和goroutine实现多阶段并发数据处理。1. 基本结构含生产者、中间处理、消费者三部分；2. 支持扇出扇入提升吞吐；3. 需正确关闭channel并用context管理生命周期；4. 应合理缓冲、避免副作用、监控性能。
在Golang中，任务流水线（Pipeline）是一种常见的并发模型，用于将数据处理过程拆分为多个阶段，每个阶段并发执行，通过channel传递数据。这种模式能有效提升程序吞吐量，尤其适用于需要多步处理大量数据的场景。
1. 基本流水线结构
一个典型的流水线由三个部分组成：生产者、中间处理阶段、消费者。各阶段通过channel连接，实现解耦和并发执行。
例如，构建一个简单的整数处理流水线：生成数字 → 平方 → 过滤偶数。
func generate(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 filterEven(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 }
使用方式：
立即学习“go语言免费学习笔记（深入）”；
nums := generate(1, 2, 3, 4, 5) squared := square(nums) even := filterEven(squared) for result := range even { println(result) // 输出: 4, 16 }2. 并发执行与扇出扇入（Fan-out/Fan-in）
为了提高处理能力，可以在某个阶段启动多个goroutine并行处理，然后合并结果。
以“平方”阶段为例，启动两个worker同时工作：
func squareFanIn(in <-chan int) <-chan int { out := make(chan int) // 启动两个worker workers := 2 for i := 0; i < workers; i++ { go func() { for n := range in { out <- n * n } }() } // 单独goroutine关闭out go func() { for i := 0; i < workers; i++ { <-in // 等待所有worker完成（简化示例） } close(out) }() return out }
更严谨的做法是使用sync.WaitGroup管理worker生命周期。
3. 正确处理关闭与资源清理
避免goroutine泄漏的关键是正确关闭channel和确保所有goroutine退出。
常见做法：
- 只由发送方关闭channel
- 接收方使用
for v := range ch自动检测关闭 - 使用context控制超时或取消
func generator(ctx context.Context, nums []int) <-chan int { out := make(chan int) go func() { defer close(out) for _, n := range nums { select { case out <- n: case <-ctx.Done(): return } } }() return out }
4. 实际应用建议
构建高效流水线需注意以下几点：
- 合理设置channel缓冲区大小，避免阻塞或内存溢出
- 中间阶段不要修改原始数据，保证无副作用
- 使用context统一管理生命周期，便于取消整个流水线
- 监控各阶段性能，识别瓶颈
基本上就这些。Golang的pipeline模型依赖channel和goroutine的简洁组合，把复杂流程变成可组合、可测试的小单元。掌握好关闭机制和并发控制，就能写出高效稳定的并发程序。
