Go pipeline中每个stage必须明确输入/输出channel类型、关闭时机和错误传播路径,否则易导致死锁或goroutine泄漏;需用context控制超时,由写入方关闭channel,合理设置buffer大小以实现背压调节。

Go 里用 channel 串联多个处理函数,不是“连上就能跑”,而是必须明确每个 stage 的输入/输出边界、关闭时机和错误传播路径——否则轻则漏数据,重则死锁或 goroutine 泄漏。
stage 函数签名必须带 context.Context 和明确的 channel 类型
硬编码 channel 类型、忽略上下文,是 pipeline 卡死的第一大原因。每个 stage 必须声明自己读什么、写什么、何时该停。
- 输入 channel 只读(
<-chan T),输出 channel 只写(chan<- U),避免类型混杂和误写 -
ctx context.Context要作为第一个参数传入,不能只在启动 goroutine 时传,而要在所有阻塞点(如http.Do、<-time.After、甚至select等待)检查ctx.Done() - 别在 stage 内部调
context.WithTimeout——超时控制必须由最外层统一决定,否则 cancel 信号无法穿透 - 示例签名:
func parseJSON(ctx context.Context, in <-chan string) <-chan map[string]interface{}
channel 关闭必须由写入方负责,且仅 close 一次
谁发数据,谁关 channel;下游靠 for range 自动退出,不是靠手动判断 ok。
- 每个 stage 启动 goroutine 后,必须用
defer close(out)—— 即使中间有 filter 导致没写任何值,也要关 - 上游提前 close(
in) 不等于下游能立刻退出:如果 stage 正在做耗时操作(比如解析大 JSON),它得自己感知ctx.Done()才能及时 return,然后 defer 才会触发 - 绝对不要在多个 goroutine 里对同一个 channel 调
close,panic: close of closed channel 会直接崩掉整个 pipeline - 常见错误:
for { v, ok := <-in; if !ok { break }—— 这种写法容易漏关、难保证 defer 执行,一律换成for v := range in
扇入(merge)和扇出(fan-out)必须显式控制,不能依赖 select 随机转发
多个输入合并或单个输出分发,不是语法糖,而是数据流拓扑的关键节点,处理不当就丢数据或打满 CPU。
- 扇入推荐写法:
func merge(cs ...<-chan int) <-chan int,为每个c启一个 goroutine:go func(c <-chan int) { for v := range c { out <- v } }(c),最后close(out) - 扇出慎用
select { case out1 <- v: ... case out2 <- v: ... }—— 它不保序、不均衡,可能让某个下游饿死;真要分发,用轮询索引或加权重标记 - 别让一个 goroutine 同时读多个 channel 并往同一个
out写:无缓冲时极易死锁;有缓冲也掩盖了背压问题 - 所有被 merge 的输入 channel,可以提前 close,
range会自动退出,不影响 merge goroutine 继续从其他 channel 读
buffer 大小不是越⼤越好,要匹配阶段间吞吐差和下游消费下限
buffer 设错,要么 OOM,要么吞吐暴跌——它不是性能开关,而是背压调节阀。
- 无缓冲 channel(
make(chan T))适合低延迟、强同步场景,但上游极易被下游拖慢;一旦某 stage 偶尔卡顿,整条 pipeline 就堵死 - 缓冲太大(如
make(chan string, 1000000))等于把数据全缓在内存里,失去流式处理意义,还可能触发 GC 毛刺 - 初始调试建议用
make(chan T, 64),上线后监控len(ch)值:若长期 >80% 容量,说明下游太慢或 buffer 不足;若常为 0,说明 buffer 过大或上游太慢 - 对不确定长度的输入(如文件逐行读取),绝不用固定大 buffer;改用动态批处理 + 显式背压反馈(比如下游返回 ack)
真正难的从来不是写几个 go func 和 ch <-,而是每个 stage 都得回答三个问题:我从哪读、往哪写、什么时候停——少一个,pipeline 就不是流水线,而是定时炸弹。


















