os.pipe不适合逐行清洗管道，因其易因缓冲和关闭时机不协调导致死锁；应使用io.pipe配合func(io.reader) io.reader链式组合，每步在独立goroutine中运行并显式close，避免阻塞与泄漏。
为什么 os.Pipe 不适合做“逐行清洗”的管道
直接用 os.Pipe 拼接多个清洗函数，很容易卡住或死锁——因为读写双方没协调好缓冲和关闭时机，尤其当某一步骤提前退出时，另一端会永远阻塞在 Read 或 Write 上。真正可行的不是“系统级管道”，而是基于 io.Reader/io.Writer 接口的组合式流处理。
用 io.Pipe 实现非阻塞、可链式调用的清洗链
io.Pipe 是 Go 标准库提供的轻量级内存管道，一端写、一端读，且自带同步机制。关键在于：每个清洗步骤应封装为 func(io.Reader) io.Reader，再通过 io.Pipe 串起来，避免 goroutine 泄漏和死锁。
- 每个清洗函数必须在自己的 goroutine 中运行，否则会阻塞上游
- 务必在 goroutine 内显式调用
pipeWriter.Close()，否则下游Read会永远等待 EOF - 错误需从 goroutine 传回主流程，不能只打印或忽略
示例：把输入按行拆解、去掉空行、转大写、再拼回带换行的字节流：
func upperCaseLineFilter(r io.Reader) io.Reader {
pr, pw := io.Pipe()
go func() {
defer pw.Close() // 必须关闭，否则下游 Read 不会结束
scanner := bufio.NewScanner(r)
for scanner.Scan() {
line := strings.TrimSpace(scanner.Text())
if line == "" {
continue
}
_, _ = pw.Write([]byte(strings.ToUpper(line) + "\n"))
}
// 注意：scanner.Err() 可选检查，但 pw.Close() 仍需执行
}()
return pr
}
如何安全地组合多个清洗步骤（比如 trim + filter + replace）
不要嵌套调用如 upperCaseLineFilter(trimLineFilter(os.Stdin))——这会导致中间 io.Reader 无法被 gc，且错误难以传递。正确做法是逐层包装，每步都新建 io.Pipe，并统一管理 goroutine 生命周期。
- 推荐用切片存清洗函数：
[]func(io.Reader) io.Reader，然后循环应用 - 每步都起独立 goroutine，输入来自上一步的
io.Reader，输出交给下一步的io.Reader - 最外层的 reader（最终结果）可直接传给
io.Copy或bufio.Scanner - 如果某步出错，应让整个链提前终止，而不是静默跳过
简化的组合逻辑示意：
func chainFilters(r io.Reader, filters ...func(io.Reader) io.Reader) io.Reader {
for _, f := range filters {
pr, pw := io.Pipe()
go func() {
defer pw.Close()
if err := pipeOne(f, r, pw); err != nil {
pw.CloseWithError(err) // 让下游 Read 返回 err
}
}()
r = pr
}
return r
}
逐行清洗时最容易被忽略的边界问题
行边界不是“有 \n 就切”，而是受 bufio.Scanner 的 MaxScanTokenSize 和换行符兼容性影响。Windows 的 \r\n、老 Mac 的 \r、超长行都会导致行为不一致。
-
scanner.Scan()默认只识别\n，\r\n会被当作普通字符；需用scanner.Split(bufio.ScanLines)并手动处理\r - 单行超过 64KB 会触发
bufio.ErrTooLong，必须显式设置scanner.Buffer - 清洗后写入时若漏掉换行符，下游可能把多行当成一行读取——尤其当最后一条记录无结尾换行时
- 不要在清洗函数里对
io.Reader做多次ReadAll，它破坏流式语义，失去“逐行”意义
实际清洗逻辑中，strings.TrimSuffix(line, "\r") 往往比依赖系统换行更可靠。
