Go 中用 channel 实现 pipeline 需满足：无缓冲 channel + 显式关闭 + range 接收；每个 stage 函数接收 in chan T、返回 out chan U，并在所有路径（含 panic defer）关闭 out；调用方控制 goroutine 并发；需监控阻塞时长与缓冲水位，确保吞吐匹配、错误可观察隔离。
Go 里怎么用 channel 实现 pipeline 链式处理
Pipeline 的本质是把数据流经多个 stage，每个 stage 做单一职责的转换或过滤，靠 chan 串起来。不是用 goroutine 包一层就叫 pipeline，关键在「无缓冲 channel + 显式关闭 + range 接收」这三点配合。
常见错误是某个 stage 忘记 close channel，导致下游 range 永远卡住；或者用带缓冲 channel 掩盖了背压问题，上游猛塞、下游来不及处理，内存暴涨。
- 每个 stage 函数接收一个
in chan T，返回一个out chan U - stage 内部必须在所有路径上
close(out)（包括 panic 后的 defer） - 上游 stage 不要自己开 goroutine 发送——由调用方控制并发更清晰
- 如果某 stage 可能丢弃数据（比如 filter），别用
select { case out 裸写，容易漏关 channel；改用 <code>if ok := sendNonBlocking(out, x); !ok { return }这类封装
示例：读文件 → 解析 JSON → 提取字段 → 写 DB
func readLines(r io.Reader) <-chan string {
out := make(chan string)
go func() {
scanner := bufio.NewScanner(r)
for scanner.Scan() {
out <- scanner.Text()
}
close(out) // 必须关
}()
return out
}为什么用 for-range 从 channel 读会 panic “send on closed channel”
这不是 pipeline 独有，而是对 channel 关闭时机理解错。panic 出现在往已关闭的 out 写数据时，但根源常在 stage 函数没处理好“上游提前关闭”或“自己提前退出”。
立即学习“go语言免费学习笔记（深入）”；
使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器，提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象，支持命名依赖等
典型场景：中间 stage 因校验失败想提前结束，但上游还在发，它又没及时退出接收循环，等 finally 关闭自己的 out 后，上游还在往这个已关的 out 里塞数据。
- 所有接收方必须用
for x := range in，不能for { x, ok := —— 后者漏掉 close 通知后的零值 - 发送方要在确认“不会再往
out发”之后才close(out)，且确保所有发送路径都覆盖（包括 error return 和 defer） - 如果 stage 需支持中断（如 ctx.Done()），用
select监听ctx.Done()并立即close(out)，但注意：此时可能有 goroutine 正在往out写，需加锁或用 sync.Once
数据清洗阶段如何安全做类型转换和空值过滤
ETL 最容易崩在脏数据上，比如 JSON 字段缺失、类型错（string 当 number 用）、编码乱码。硬写 json.Unmarshal + interface{} 断言，出错就 panic，根本扛不住线上流量。
- 用结构体 +
json.Number或自定义UnmarshalJSON方法，把解析逻辑收口，错误统一转成error返回，不要 recover - 空值过滤别写
if v == nil，Go 里nil对 slice/map/func/chan 有效，但对 struct、int、string 无效；用指针字段 +if v != nil && *v != "" - 时间解析别直接
time.Parse，先用strings.TrimSpace去首尾空格，再判断是否为空字符串，否则Parse("", ...)panic - 数值转换优先用
strconv.ParseInt(s, 10, 64)而非json.Number.Int64()，后者对超大数会溢出返回 0 且不报错
示例：清洗用户年龄字段
type User struct {
Age *int `json:"age"`
}
// 清洗函数返回 (cleaned *User, err error)，不修改原数据高吞吐下 pipeline 性能瓶颈在哪，怎么定位
瓶颈通常不在 CPU，而在 channel 阻塞、GC 压力、或系统调用（如文件读、DB 写）。用 go tool pprof 看 runtime.gopark 占比高，基本就是 channel 等待；看 runtime.mallocgc 高，说明小对象分配太频繁。
- 避免在 pipeline 中频繁创建 map/slice——复用
sync.Pool，尤其 JSON 解析后的临时 struct - IO 密集型 stage（如写 Kafka）别用单个 goroutine 塞满 channel，改用 worker pool：启动固定数量 goroutine 从 channel 拿数据批量提交
- 不要让 pipeline 最后一环（如 DB 写入）变成单点瓶颈——它应该消费速度 ≥ 上游生产速度，否则 channel 缓冲区堆满，上游 goroutine 全卡住
- 监控每 stage 的 channel len / cap 比值，持续 > 0.8 就说明下游慢了；用
runtime.ReadMemStats定期打点 GC 次数和 pause 时间
真正难的不是搭起 pipeline，是让每个 stage 的吞吐能力匹配，且错误能被观测、被隔离、不拖垮整条链。实际跑起来后，第一个要盯的永远是 channel 的阻塞时长和缓冲区水位。
