Go io.Pipe 如何连接生产者和消费者：阻塞传递与 CloseWithError

做流式导出时，生产者往往一边生成数据，消费者一边压缩、上传或写入文件。Go 的 io.Pipe 可以把这两段代码直接接起来，但它不是带缓存的队列：生产者写得太快，Write 就会等待读端；生产者失败，也必须主动把错误传到读端。

io.Pipe 适合把两个同步的 io.Reader/io.Writer 阶段连成流。记住它没有内部缓冲，并用 CloseWithError 收口失败路径，代码才不会留下悬挂的 goroutine。

先确定：这里需要的是同步流，不是缓存队列

io.Pipe() 返回 *io.PipeReader 和 *io.PipeWriter。写入的数据会从 PipeWriter 直接交给 PipeReader，中间没有一块由管道管理的缓冲区。官方文档还特别说明，一次 Write 可能需要等待多个 Read 才能完整消费。

这让它很适合“生成一段、处理一段”的链路，例如把 CSV 导出直接送进压缩器；但不适合把突发流量暂存起来。需要解耦生产速度和消费速度时，应另行设计带容量的 channel 或消息系统。

Write 为什么会停住：两端是一条有反压的调用链

下面这个小例子故意把生产者和消费者放在两端。PipeWriter 的 Write 只有在 PipeReader 的 Read 接住数据后才会继续，因此“写入函数返回”本身就是消费者已经取得这批数据的一个检查点。

package main

import (
    "fmt"
    "io"
    "log"
    "strings"
)

func main() {
    reader, writer := io.Pipe()

    go func() {
        defer writer.Close()
        for _, part := range []string{"header\n", "row-1\n", "row-2\n"} {
            if _, err := writer.Write([]byte(part)); err != nil {
                log.Println("写入失败:", err)
                return
            }
        }
    }()

    var out strings.Builder
    if _, err := io.Copy(&out, reader); err != nil {
        log.Fatal(err)
    }
    fmt.Print(out.String())
}

这里的控制流只有一条：生产者调用 Write，消费者的 Read 接收，io.Copy 读到 EOF 后返回。生产者用 defer writer.Close() 表示正常结束，否则消费者会一直等下去。

用一个可观察的检查点验证阻塞关系

调试时可以在 Write 前后打印日志：如果前一条日志出现、后一条迟迟不出现，而消费者没有开始读，问题不是锁，而是 io.Pipe 正在施加反压。不要给管道“加一个隐形缓存”的错觉，数据不会在 PipeWriter 内排队。

生产者失败时，CloseWithError 要把原因送到读端

正常完成使用 Close，消费者最终读到 EOF。若生产者中途失败，仅仅返回 goroutine 会让消费者继续等待；应该调用 writer.CloseWithError(err)。随后读端的 Read 或 io.Copy 会观察到这个错误。

reader, writer := io.Pipe()

go func() {
    if _, err := writer.Write([]byte("partial\n")); err != nil {
        return
    }
    writer.CloseWithError(fmt.Errorf("导出记录校验失败"))
}()

_, err := io.Copy(dst, reader)
if err != nil {
    // err 包含生产者通过 CloseWithError 传来的原因
    return fmt.Errorf("消费导出流: %w", err)
}

这条路径的关键不是“把错误打印出来”，而是让消费者停止把结果当成完整文件。尤其是已经收到 partial 的情况下，调用方仍要依据 io.Copy 的错误决定是否删除临时文件或回滚上传。

退出收口：谁创建，谁负责让另一端结束

最容易泄漏的是这三种情况：消费者提前返回却没关闭读端；生产者遇到错误只记录日志；成功路径忘记关闭写端。可以按下面的责任划分检查：

• 消费者主动取消：调用 reader.CloseWithError(err)，让正在 Write 的生产者尽快返回。
• 生产者正常完成：调用 writer.Close()，让消费者看到 EOF。
• 生产者失败：调用 writer.CloseWithError(err)，让消费者拿到真实原因。

如果链路还要接入请求上下文，应在生产者循环中同时检查取消信号；不要指望关闭 HTTP 响应就自动结束一个仍在写 PipeWriter 的 goroutine。

常见误区与边界

把 io.Pipe 当成带容量的 channel

它没有内部缓冲。想预存数据，选有明确容量的 channel；想把数据落盘，使用文件或专门的临时存储。io.Pipe 的价值是同步连接接口，而不是吸收峰值。

只检查 Write，不检查 io.Copy

Write 成功只能说明这部分数据被读端接收，不能证明整个结果成功。最终状态要看消费者的 io.Copy 返回值，以及生产者是否用 Close 或 CloseWithError 明确结束。

在同一个 goroutine 里先 Write 再 Read

这种顺序会互相等待。至少要让生产者在 goroutine 中运行，或者使用已经存在的并行消费者，否则第一处 Write 没有读端接收就不会返回。

把一条 io.Pipe 链路验收完整

先用小数据验证正常路径：消费者应读到完整内容，io.Copy 返回 nil。再让生产者在写入部分内容后调用 CloseWithError，确认消费者收到错误而不是把部分文件当成功。最后模拟消费者提前取消，确认生产者的 Write 能返回并且 goroutine 数量不会持续上涨。

io.Pipe 的判断标准很朴素：两端是否确实需要同步传递？成功和失败是否都能到达另一端？每一条路径是否都有关闭动作？这三个问题都能回答清楚时，它就是一件简洁而可靠的流连接工具。

相关问题

io.Pipe 会缓存已经写入的数据吗？

不会。官方文档明确说明它没有内部缓冲，写入与读取会同步匹配。

为什么消费者会一直卡在 io.Copy？

通常是生产者没有调用 writer.Close()，或者错误路径没有调用 writer.CloseWithError(err)，读端因此没有收到结束信号。

CloseWithError 之后还应该继续 Write 吗？

不应该。关闭表示这条管道进入结束状态，生产者应立即收口并返回，把后续清理交给调用方。