每个清洗阶段必须是独立函数，接收输入 channel 并返回输出 channel，内部需处理关闭信号、避免无限阻塞、及时退出 goroutine，确保 pipeline 可观测、可终止、不卡死。
Go 语言构建高吞吐量数据清洗项目，核心不在“堆 goroutine”，而在于用好 channel 控制流、用对并发粒度、并让每个 stage 可观测可终止。盲目开协程反而导致 goroutine 泄漏、内存暴涨、死锁频发。
怎么写一个不会卡死的 pipeline stage
每个清洗阶段必须是独立函数，接收 、返回 <code>，且内部所有退出路径（包括 panic）都需 <code>close(out)。常见错误是只在正常流程 close，但 error return 或 defer 里漏掉。
- 必须用
for x := range in消费输入，不能用for { x, ok := —— 后者会在 channel 关闭后仍读一次零值 - 发送前要确保
out还没被关：若某条数据校验失败想跳过，别直接out ，改用带判断的封装，比如 <code>sendIfOpen(out, x) - stage 内部不要自己
go func() { ... }()启动发送逻辑；启动时机应由调用方统一控制，否则并发不可控、调试困难
缓冲 channel 该不该加、加多少
无缓冲 channel 能天然实现背压，但一旦某个 stage 处理变慢（比如调外部 API），上游就会阻塞。是否加缓冲，取决于清洗耗时的波动性。
- 纯 CPU 计算型清洗（如字符串 trim、正则替换）→ 通常不加缓冲，靠背压自动限速更稳
- IO 密集型（如查 Redis 校验手机号、调 HTTP 接口）→ 建议
make(chan T, 100)，防瞬时积压拖垮上游 - 缓冲大小不是拍脑袋定的：先设为 100，上线后通过 pprof 查看
runtime.ReadMemStats中Mallocs和HeapInuse趋势，再结合日志中 channel 阻塞超时告警调整
怎么防止 goroutine 泄漏和数据丢失
泄漏和丢失本质都是生命周期没对齐：主 goroutine 提前退出，或子 goroutine 卡在 channel 上永远等不到关闭信号。
立即学习“go语言免费学习笔记（深入）”；
- 绝不能裸写
go cleanStage1(in, out)就不管 —— 必须配sync.WaitGroup或用for range out驱动消费，靠close(out)触发退出 - 如果清洗链中有 filter 类 stage（可能丢弃部分数据），仍要保证它主动退出，否则上游
range in会一直等下去；可在 stage 结束时加log.Printf("stage1 done, processed %d items")辅助定位卡点 - 所有涉及网络/文件 IO 的 stage，必须设超时：比如用
context.WithTimeout包裹 HTTP 调用，超时后 clean 函数应立即 return 并 close(out)
真实场景下怎么观察 pipeline 是否健康
线上跑着的 pipeline 不是黑盒，得能快速判断哪一环拖后腿、有没有堆积、是否已死锁。
- 每个 stage 启动时注册 pprof：
http.ListenAndServe(":6060", nil)，用go tool pprof http://localhost:6060/debug/pprof/goroutine?debug=2看 goroutine 堆栈，重点查卡在chan send或chan recv的数量 - 用
runtime.ReadMemStats定期采样，关注HeapAlloc增长速率 —— 若持续上涨且不回落，大概率是某 stage 忘关out，下游 range 永远等不到 EOF - 给关键 channel 加水位监控：封装一个带计数器的 channel wrapper，在每次
和 <code> 前后更新指标，暴露到 Prometheus，看 buffer length 是否长期 >80%
真正难的不是写出能跑的 pipeline，而是让每个 stage 在异常（panic、超时、上游提前关闭）下仍能干净退出、不残留 goroutine、不泄露内存。这些边界情况，往往在线上流量突增或依赖服务抖动时集中爆发。
