Go io.Pipe 怎样把生产者和消费者连接成流

先给结论：io.Pipe() 返回一对 *io.PipeReader 和 *io.PipeWriter。把生产者放进 goroutine，让它写入 PipeWriter；消费者直接读取 PipeReader，就能把两段只认识 io.Writer 和 io.Reader 的代码接成一条流。它没有内部缓冲，写入会等待读端消费，所以天然带有背压。

官方地址：https://pkg.go.dev/io

先把 io.Pipe 的连接语义讲清楚

我第一次把生成器接到下游编码器时，最容易误判的是“写入成功”这四个字。io.Pipe 的 Write 并不会把数据放进一个隐藏缓冲区后马上返回；它会等待一个或多个 Read 把这次写入的数据消费完。读端如果暂时没有推进，生产者就会停在写入处。

因此它适合把“边生成边消费”的两个组件接起来，而不适合把它当作带容量的消息队列。官方实现用同步通道匹配读写，数据直接从写端复制到对应读端；这也是图中要强调的核心边界。

图1：io.Pipe 的两个端点、无内部缓冲边界与读写契约关系。

让生产者和消费者同时推进

最小写法是先创建管道，再启动生产 goroutine，主 goroutine 或另一个 worker 负责读取。下面的例子把三段文本逐块写入，读端用 io.ReadAll 收集结果；代码中的注释保留了关闭和错误处理的职责。

package main

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

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

    go func() {
        // 生产者只依赖 io.Writer，不需要知道消费者如何处理数据。
        chunks := []string{"header|", "body|", "tail"}
        for _, chunk := range chunks {
            if _, err := io.WriteString(writer, chunk); err != nil {
                // 读端提前关闭时，立刻结束生产 goroutine。
                log.Printf("write pipe: %v", err)
                return
            }
        }
        // 正常关闭让读端在数据消费完后得到 io.EOF。
        _ = writer.Close()
    }()

    // 消费者只依赖 io.Reader，可以替换成编码器或上传请求。
    data, err := io.ReadAll(reader)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(string(data))
}

这里的顺序不是“先全部写完、再全部读取”：生产 goroutine 会在每次写入处与读端同步。io.ReadAll 只是为了让示例容易观察结果；真实消费方可以把 reader 交给解码器、压缩器或 HTTP 请求体，只要它们接受 io.Reader。

收尾和失败要用不同的信号

正常完成时，写端调用 Close()，读端会在已写数据读完后收到 io.EOF。如果生产端遇到错误，直接 Close() 会把“正常结束”和“生产失败”混在一起；应改用 CloseWithError(err)。它不会覆盖已经记录的更早错误，因此应由负责生产的那一侧尽早发送。

func stream(reader *io.PipeReader, writer *io.PipeWriter) error {
    go func() {
        // 业务生产失败时，把错误原样传给读端。
        if err := produce(writer); err != nil {
            _ = writer.CloseWithError(err)
            return
        }
        // 没有错误才发送正常结束信号。
        _ = writer.Close()
    }()

    _, err := io.Copy(io.Discard, reader)
    // EOF 表示正常结束，其他错误应继续向上返回。
    if err == io.EOF {
        return nil
    }
    return err
}

produce 只需要遵守“成功返回、失败返回 error”的约定即可。另一个常见边界是读端提前退出：它应该关闭 PipeReader，这样正在阻塞的写端会获得 io.ErrClosedPipe 或读端提供的错误，生产 goroutine 才有机会退出。

图2：正常 EOF、CloseWithError 和读端提前关闭之间的错误传播边界。

哪些场景适合，哪些场景要换思路

如果上游正在生成归档、编码或序列化结果，下游可以边读边上传、边解码或边写入另一个 io.Writer，io.Pipe 能减少“先生成完整临时文件”的等待和额外存储。它的同步特性还会把下游处理速度反馈给上游，避免生产端无上限堆内存。

但如果业务需要让读者随时回看旧数据、允许生产端独立突发写入，或者需要按偏移量随机读取，就不要把 io.Pipe 当缓存使用。此时应在两端之间明确加入有容量的队列、文件或其他可回读存储，并单独设计取消和容量策略。

简短问答

io.Pipe 为什么必须启动 goroutine？

因为读写是同步匹配的。如果在同一个 goroutine 里先调用会阻塞的 Write，又没有并行的 Read，程序就会互相等待。

读端什么时候得到 EOF？

写端正常调用 Close() 后，读端先消费完已经写入的数据，再得到 io.EOF；不要只依赖 goroutine 自然结束来表示流结束。

CloseWithError 会覆盖之前的错误吗？

不会。PipeReader 和 PipeWriter 的文档都说明，关闭方法不会覆盖已经保存的错误，所以要在错误发生处及时发送。

把 io.Pipe 记成“无缓冲、同步、可传递关闭错误的 Reader/Writer 适配器”，生产者和消费者的职责就能保持清楚：生产者负责写完或报告失败，消费者负责持续读取并处理结束状态。