<p>Go流式处理核心是chan与context.Context组合：数据经goroutine链式流转，每个stage需独立goroutine、select监听ctx.Done()、defer close(out)，输入用<-chan T、输出用chan<- T。</p>
流式处理的核心不是“框架”，而是 chan 与 context.Context 的组合用法
Go 里没有所谓“开箱即用的流式数据处理框架”，强行套用 Java 或 Rust 的流式 API（如 map/filter/reduce 链式调用）反而会破坏 Go 的并发模型。真正的流式处理在 Go 中体现为：数据从一个 chan 流入，经若干 goroutine 处理后，再流入下一个 chan，全程受 context.Context 控制生命周期。
常见错误是把所有 stage 写成同步函数调用，或忽略 channel 关闭时机，导致 goroutine 泄漏。正确做法是每个 stage 都启动独立 goroutine，并在 ctx.Done() 时主动退出、关闭输出 channel。
- 输入 channel 类型建议统一为
<-chan T（只读），输出为chan<- T（只写），避免误写 - 每个 stage 必须用
defer close(out)确保输出 channel 最终关闭，否则下游range会永远阻塞 - 不要在 stage 内部直接
close(in)—— 输入 channel 应由上游控制
如何实现带背压的流式 pipeline（例如日志解析 + 过滤 + 上报）
背压不是靠缓冲区大小硬扛，而是靠 channel 的阻塞特性天然实现。关键在于：所有 stage 的 input channel 都不带缓冲（make(chan T)），这样当下游处理慢时，上游写操作会自动阻塞，形成反向压力传导。
示例场景：从 stdin 读行 → 解析 JSON → 过滤 error 日志 → 发送到 HTTP 接口：
立即学习“go语言免费学习笔记（深入）”；
使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器，提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象，支持命名依赖等
func parseJSONLines(ctx context.Context, in <-chan string) <-chan map[string]interface{} {
out := make(chan map[string]interface{})
go func() {
defer close(out)
for {
select {
case line, ok := <-in:
if !ok {
return
}
var v map[string]interface{}
if err := json.Unmarshal([]byte(line), &v); err == nil {
select {
case out <- v:
case <-ctx.Done():
return
}
}
case <-ctx.Done():
return
}
}
}()
return out
}- 所有
select必须包含ctx.Done()分支，且放在第一顺位，防止 goroutine 卡死 - HTTP 上报 stage 若使用
http.Do，需设 timeout 并用ctx.WithTimeout包裹 client，否则单个失败请求可能拖垮整条 pipeline - 若需跳过脏数据而非 panic，用
continue而非return，否则当前 stage 提前退出，后续数据丢失
sync.Pool 在流式处理中基本没用，别乱用
有人看到“高吞吐”就下意识加 sync.Pool 缓存结构体，但在典型流式 pipeline 中，每个 item 生命周期短、类型固定、分配频次未必高，盲目引入 sync.Pool 反而增加 GC 压力和逻辑复杂度。
真正该复用的是：长连接（如 *http.Client）、编解码器（如 json.Decoder）、或大 buffer（如 bytes.Buffer）。这些对象初始化开销大，且可在 stage 内复用。
-
json.Decoder可复用：传入bytes.NewReader每次重置，比反复 new 更快 - 不要对
map[string]interface{}或struct{}做sync.Pool—— Go 1.22+ 的逃逸分析已足够智能 - 若真要用 pool，务必保证
Put前清空字段（尤其 slice 和 map），否则引发数据污染
调试流式 pipeline 的三个必查点
流式程序最难 debug 的不是逻辑错，而是“数据卡在哪”。一旦 pipeline 停滞，优先检查以下三项：
- 某个 stage 是否忘了
defer close(out)？用go tool trace查看 goroutine 是否卡在chan send - 是否在某处用了
range in而非for { select { case ... }}？前者会在in关闭后退出，但若上游未关 channel，循环永不启动 - context 是否被意外 cancel？比如父 context 是
http.Request.Context()，但 handler 已返回，子 pipeline 却还在跑
最易被忽略的是：多个 stage 共享同一个 context.Background()，导致无法优雅停止。实际应由最外层传入带 timeout 或 cancel 的 context，并逐级传递。
