京公网安备11010502063487号用 channel 串起来的流水线写起来简单,但有三个实际问题:消费者提前走了怎么办、并行之后顺序怎么办、缓冲区开多大。这篇用真实运行回答,并量并行 worker 数的收益。
没有取消机制:100 条流水线各只读 3 个元素就丢弃 → goroutine 数 1 → 101 读 3 个后 cancel():同样 100 条两级流水线 → goroutine 数 101 → 101(起点 101 里包含上一组已经泄漏的 100 个,这一组没有新增)
每一级发送都只用 out <- v,消费者读了 3 个就走,上游 goroutine 永远卡在发送上,100 条流水线泄漏 100 个。每一级把发送和 ctx.Done() 放进同一个 select,cancel() 之后全部退出。
4 个 worker 并行处理 0..1999:收到 2000 个,元素齐全=true,输出里“比前一个小”的位置有 794 处(顺序被打乱)
4 个 worker 并行处理、再合并,2000 个元素全部收到,但顺序不再是输入的顺序(794 处逆序)。需要顺序就要带序号并在合并后排序。
第二级 worker 数 = 1:处理 2 万个元素用时 596 ms 第二级 worker 数 = 2:处理 2 万个元素用时 320 ms 第二级 worker 数 = 4:处理 2 万个元素用时 176 ms 第二级 worker 数 = 8:处理 2 万个元素用时 114 ms
瓶颈在第二级且是 CPU 计算时,1、2、4、8 个 worker 分别用 596、320、176、114 ms。到 8 个时没有 8 倍的加速,我推测是生成和合并阶段的 goroutine 也在抢 CPU,没有验证。
三级流水线,每级 channel 容量 0:100 万个元素 594 ms, 594 ns/个 三级流水线,每级 channel 容量 1:100 万个元素 419 ms, 419 ns/个 三级流水线,每级 channel 容量 16:100 万个元素 229 ms, 229 ns/个 三级流水线,每级 channel 容量 256:100 万个元素 129 ms, 129 ns/个
三级几乎不做事的流水线,容量 0 时每个元素 594 ns,容量 256 时 129 ns,因为级间交接不用每次都等对方。但缓冲区只是在摊薄同步成本,并不改变整体的吞吐上限,而且会让背压来得更晚。