io.Pipe 不适合做计算管道，因其读写必须严格同步，任一环节阻塞即导致整条管道卡死；应改用 chan + goroutine 实现解耦的流式处理。
为什么 io.Pipe 不适合做计算管道？
直接用 io.Pipe 拼接多个处理步骤，容易卡死——因为读写两端必须严格同步，任一环节阻塞（比如某个处理器没及时读/写），整个管道就挂住。这不是流式计算要的效果，而是 I/O 协调问题。
真正需要的是“数据推着走”，每个阶段只关心输入和输出，不感知上下游状态。所以得靠 chan + goroutine 手动调度，让每个处理器在自己的 goroutine 里运行，彼此解耦。
- 用
io.Pipe写流式转换函数，90% 的失败都源于 goroutine 启动顺序或缓冲区大小误判 - 推荐统一用
func( 类型签名，明确输入输出方向，避免反向阻塞 - 所有中间 channel 必须带缓冲（哪怕
1），否则第一个处理器产出后就等着下一个来读，链式立刻断掉
如何定义可组合的流式处理函数？
关键不是“怎么写单个函数”，而是“怎么让它们能无缝拼接”。统一签名是前提：func( 这类类型，既表达语义（只读输入、只写输出），又支持类型推导和链式调用。
示例：一个过滤偶数的处理器
立即学习“go语言免费学习笔记（深入）”；
func evenFilter(in <-chan int) <-chan int {
out := make(chan int, 1)
go func() {
defer close(out)
for v := range in {
if v%2 == 0 {
out <- v
}
}
}()
return out
}- 必须在 goroutine 里启动循环，否则调用即阻塞
- channel 缓冲至少为
1，否则out 可能永久等待下游消费 - 务必
defer close(out)，否则下游range永远等不到 EOF - 不要在函数内关闭
in，那是上游责任；也不要往in写，它只是只读
链式调用时怎么避免 goroutine 泄漏？
每加一层处理器就启一个 goroutine，如果某层 panic 或提前退出，后续 goroutine 可能永远收不到输入或无法关闭输出 channel，导致泄漏。
最简方案：用 context.Context 控制生命周期，所有 goroutine 监听 ctx.Done() 并清理。
func squareWithContext(ctx context.Context, in <-chan int) <-chan int {
out := make(chan int, 1)
go func() {
defer close(out)
for {
select {
case v, ok := <-in:
if !ok {
return
}
select {
case out <- v * v:
case <-ctx.Done():
return
}
case <-ctx.Done():
return
}
}
}()
return out
}- 不能只依赖
in关闭来退出，必须响应ctx.Done() - 向
out写入前也要 select，防止 ctx 已取消还强行发数据导致 panic - 实际使用时，建议顶层传入带 timeout 的 context，例如
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
真实场景下怎么处理错误传递？
纯 channel 链无法传递 error，因为 类型不包含错误信息。常见做法是把 error 和数据一起打包，或者另开一个 error channel。
推荐方案：返回 ，其中 <code>Result 是结构体：
type Result[T any] struct {
Value T
Err error
}
<p>func parseJSON(in <-chan string) <-chan Result[map[string]interface{}] {
out := make(chan Result[map[string]interface{}], 1)
go func() {
defer close(out)
for s := range in {
var v map[string]interface{}
if err := json.Unmarshal([]byte(s), &v); err != nil {
out <- Result[map[string]interface{}]{Err: err}
continue
}
out <- Result[map[string]interface{}]{Value: v}
}
}()
return out
}- 下游必须检查每个
Result.Err，不能假设Value总有效 - 一旦某步出错，是否继续处理后续输入，由业务决定——有些场景要跳过，有些要终止整条流水线
- 如果要用 error channel 分离，注意两个 channel 的生命周期必须对齐，否则容易出现“err 到了但 data 没到”或反过来
链式流式处理真正的难点不在语法，而在 channel 生命周期管理和错误传播路径的设计。写完一个处理器不难，难的是确保十层嵌套后，cancel、error、close 都能按预期传导，而不是静默卡死或 goroutine 积压。
