Go pipeline 用 done 信号回收中间 goroutine 的实践示例

Go pipeline 最容易被忽略的坑，不是阶段怎么串起来，而是下游提前停止消费后，上游还卡在发送操作里。解决办法是让调用方持有一个只关闭一次的 done channel，每个阶段在发送和接收处都监听它；退出前关闭自己创建的输出 channel，fan-in 再用 sync.WaitGroup 等所有转发 goroutine 收口。

识别提前停止消费留下的阻塞点

一个典型 pipeline 有生成、变换和合并三个阶段。正常情况下，下游持续读取，上游的 out 最终会完成；但如果调用方只取到第一个结果就返回，后续阶段可能继续计算并发送。无缓冲 channel 会把 goroutine 卡在 send 上，有缓冲 channel 也只是把问题推迟到缓冲区写满。

因此，关闭结果 channel 并不能自动通知所有生产者。接收方也不应该擅自关闭上游的 channel，否则发送方继续写入时会触发 panic。取消信号和数据 channel 要分工：前者表达“不要再做”，后者表达“数据流结束”。

设计 done 广播信号和 channel 所有权

用一个空结构体 channel 作为广播开关即可。关闭 channel 后，所有接收它的 case 都会立即变为可选，不需要为每个 goroutine 发送一条消息。约定也要先写清：创建 pipeline 的调用方负责关闭 done，阶段函数负责关闭自己的 out，任何阶段都不关闭别人的 channel。

下面的 gen 和 square 是两个可取消阶段。示例中的代码是正文说明用的最小实现，重点在退出路径，而不是把所有业务处理塞进一个函数。

func gen(done 

把取消分支接入每次发送和接收

上面的关键不是“有一个 done channel”，而是每个可能阻塞的位置都接入取消分支。读取输入时监听 in 与 done；发送输出时再次监听 out 与 done。只在循环外判断一次 done 不够，因为判断结束后，下一次 send 仍可能卡住。

调用方可以只消费一个结果，然后关闭 done。关闭动作要由唯一责任方执行一次；多个分支都可能触发取消时，应把关闭动作收敛到调用方，或者用 sync.Once 包起来，避免重复关闭导致 panic。

func main() {
    done := make(chan struct{})
    first := square(done, gen(done, 1, 2, 3, 4, 5))

    value, ok := 

如果某个阶段还要调用数据库、HTTP 客户端或文件读取接口，done 只能取消它自己的 channel 等待；底层依赖若支持 context.Context，应把请求上下文继续向下传递，避免 goroutine 已退出但外部 I/O 仍在运行。

用 WaitGroup 收口合并阶段

fan-in 会为每个输入启动一个转发 goroutine。关闭合并输出前，必须确保这些 goroutine 不会再发送；否则既可能丢失收口，也可能出现向已关闭 channel 发送的 panic。WaitGroup 只负责等待，不负责取消，所以它要和 done 同时使用。

func merge(done 

检查关闭时机和 context 边界

落地前可以按下面的清单检查：

如果取消信号还要携带超时、错误原因或跨越 HTTP/RPC 边界，优先使用 context.WithCancel、WithTimeout 或带 cause 的变体；如果只是在一个本地 pipeline 内广播“停止”，done chan struct{} 更直观。无论选哪种方式，资源释放都要放在 goroutine 的退出路径上。

相关问题

为什么给 done 发送一个值不如 close(done)？

发送一个值通常只能唤醒一个接收者，而关闭 channel 会让所有接收者同时观察到完成信号，更适合多个 pipeline 阶段广播取消。

done 关闭后还需要关闭数据 channel 吗？

需要。done 表达取消原因，数据 channel 的关闭表达数据流结束；每个阶段仍应在退出路径关闭自己创建的输出。

什么时候应该直接传 context.Context？

当阶段还要调用支持 context 的 I/O、需要截止时间或要向上游携带取消原因时使用 context；纯本地、单一 pipeline 的停止信号可以保留 done。