直接用 goroutine 链式调用卡住吞吐的根本原因是无缓冲 channel 导致上游被下游慢处理阻塞；须用带缓冲 channel、worker pool 控制并发、每个 stage 支持 context.Context 取消与超时。
为什么直接用 goroutine 链式调用反而卡住吞吐
很多人以为“多开 goroutine 就等于高吞吐”，结果在 processData 里直接 go func() { ... }() 启动下游处理，发现 channel 阻塞、CPU 利用率低、延迟飙升。根本原因是：无缓冲 channel 在写入时会阻塞 sender，而下游处理慢（比如涉及网络 I/O 或解析）时，上游 producer 立刻被卡死，整条流水线停摆。
- 必须用带缓冲的
out := make(chan int, 10)，缓冲大小要匹配下游处理速率——太小易满，太大浪费内存且掩盖背压问题 - 每个 stage 的 goroutine 必须
defer close(out)，否则消费者永远等不到 EOF - 别在 pipeline 中混用同步逻辑（如未设超时的
http.Get），它会让整个 goroutine 卡在 syscall，拖垮 P 调度器
如何用 worker pool 控制并发数，避免 goroutine 泛滥
流式处理不是“每个数据启一个 goroutine”，而是复用固定数量的 worker 消费任务队列。盲目让 processData 每次都起新 goroutine，跑几万条数据就可能创建上万个 goroutine，调度开销反超收益。
- worker 数量建议设为
runtime.GOMAXPROCS(0)的 2–4 倍；若 stage 含 HTTP 调用，可略放大，但需配合http.Transport.MaxIdleConns调优 - 任务 channel 必须带缓冲：
jobs := make(chan Task, 1000)，否则生产者一快就堵死 - 每个 worker 要主动检查
ctx.Done()，支持统一取消；panic 必须 recover，否则一个 panic 崩掉整个 pool
什么时候该加 context.Context，而不是只靠 channel 关闭
仅靠 close(out) 只能表达“数据发完了”，但无法应对超时、取消、错误中断等真实场景。比如某个 stage 因下游服务不可用卡住，上游还在不停生产，缓冲区打满后所有 goroutine 全部阻塞。
初始化并管理阿里云SDK Go 客户端，涵盖 sync.Once 单例、goroutine 安全、endpoint 与 region 配置、VPC 端点及调试。
- 每个 stage 函数签名应为
func stage(in - 内部用
select { case 替代直写 channel - 主流程用
errgroup.Group启动所有 stage，并统一等待或传播 cancel —— 不用手动管理 done channel
bufio.Reader 加在哪一级才真正提升吞吐
流式处理常对接 TCP 连接、文件或标准输入，但很多人在 pipeline 最外层套一层 bufio.NewReader 就以为优化完成。实际上，buffering 的收益取决于数据粒度和协议特征。
立即学习“go语言免费学习笔记（深入）”；
- 服务端：在
net.Conn接收侧加bufio.NewReaderSize(conn, 8192)，能大幅减少read系统调用次数，尤其对小包协议（如 MQTT、自定义二进制帧）效果明显 - 客户端：若 pipeline 消费的是 HTTP 响应体，应在
resp.Body上套bufio.NewReader，而非在 stage 内部反复io.Copy或json.Decoder直读 - 别在流式响应（如 SSE、chunked transfer）的 writer 侧滥用
bufio.NewWriter，缓冲不及时 flush 会导致首字节延迟，破坏实时性
实际部署时最容易被忽略的是背压信号的传递路径：从最下游 consumer 的消费速率，要能逐级反向影响 upstream producer 的生成节奏。这不单靠 channel 缓冲，更依赖 context 超时、worker pool 饱和反馈、以及连接层（如 net.Listener 的 SO_BACKLOG）的协同限流。漏掉任何一环，吞吐都会在某个隐匿节点突然坍塌。
