Go io.Pipe 怎么连接生产者与消费者

Go 里的 io.Pipe 专门用来连接“只会写入”的生产者和“只会读取”的消费者。调用 pr, pw := io.Pipe() 后，把 pw 交给生产端，把 pr 交给消费端；两端必须并发工作，生产结束时还要关闭写端。官方文档位于 https://pkg.go.dev/io#Pipe。

io.Pipe 为什么能连接生产者与消费者

io.Pipe() 返回相互连接的 *io.PipeReader 和 *io.PipeWriter。生产者只依赖 io.Writer，消费者只依赖 io.Reader，因此双方不需要知道彼此的实现细节。常见用途包括边生成边上传、边压缩边发送，以及把编码器输出直接交给解析器。

这个管道没有内部缓冲。一次 Write 只有在相应数据被 Read 完整取走后才返回，所以慢消费者会自然限制快生产者，这就是背压。它能避免无限堆积数据，但也意味着不能先在同一个 goroutine 里写完再读，否则第一次写入就可能永久等待。

最小可用写法：生产端放进 goroutine

下面让生产者逐行写入，消费者使用 io.Copy 把数据复制到标准输出。写端位于 goroutine 中，读端在主 goroutine 中运行，两侧能够同时推进。

package main

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

func main() {
    pr, pw := io.Pipe()

    go func() {
        // 生产端结束时必须关闭写端，让读端最终收到 EOF。
        defer pw.Close()

        for i := 1; i 

io.Copy 读到正常的 EOF 时会返回 err == nil。因此，生产者成功完成后关闭 PipeWriter，正好表达“没有更多数据”；消费者不需要把 EOF 当成业务错误。

生产失败时用 CloseWithError 传递原因

只调用 Close 会让消费者看到正常结束。如果生产过程失败，应把原始错误交给 CloseWithError。消费端会先读完已经交付的数据，随后在下一次读取时收到这个错误，便于区分“完整结束”和“中途失败”。

func produce(pw *io.PipeWriter) {
    // 模拟生产器先成功输出一段数据。
    if _, err := io.WriteString(pw, "header\n"); err != nil {
        return
    }

    err := fmt.Errorf("生成正文失败")
    // 把生产错误传到读端；不要再额外调用 Close 覆盖语义。
    _ = pw.CloseWithError(err)
}

同样地，如果消费者决定提前停止，应关闭 PipeReader，必要时使用 CloseWithError。这样仍在等待的写操作会被唤醒并返回错误，生产者 goroutine 才不会悄悄泄漏。

关闭顺序与错误边界

Read、Write 和 Close 可以并行调用；多个并行读取或多个并行写入也会被管道顺序协调。不过，这不意味着业务数据自动具备消息边界。一次大写入可能由多次读取完成，所以协议仍应使用换行、长度前缀或其他明确的分帧方式。

最常见的阻塞与泄漏问题

先写后读：在同一个 goroutine 中先执行 pw.Write，准备写完后再读取。由于写入等待读取，代码停在第一步。解决办法是让读写两端并发运行。

忘记关闭写端：生产者已经返回，但消费者仍在等待下一批数据。确保所有成功路径最终执行 Close，所有失败路径执行 CloseWithError。

消费者提前退出却不关读端：生产者可能永远阻塞在写入。消费端决定停止时应关闭 PipeReader，让写端及时获得错误。

把 io.Pipe 当队列：它没有容量，也不保存待处理消息。若需要让生产者短暂领先消费者，应使用带缓冲的 channel、自建队列或其他存储，而不是期待 io.Pipe 自动缓存。

选择 io.Pipe 的判断标准

当上游 API 只接受 io.Writer、下游 API 只接受 io.Reader，并且数据适合边生成边消费时，io.Pipe 很合适。若数据量很小且必须反复读取，bytes.Buffer 更简单；若传递的是结构化任务而非字节流，channel 通常更自然。

落地前检查五项：读写是否并发、成功是否关闭写端、失败是否保留原始错误、消费者提前退出时是否关闭读端、数据协议是否有明确边界。满足这些条件后，io.Pipe 就能成为一个内存占用稳定、接口组合清晰的流式连接器。

相关问题

io.Pipe 内部有多大缓冲区？

没有内部缓冲。数据直接从写操作交给一个或多个读操作。

可以让多个 goroutine 同时写 PipeWriter 吗？

标准库允许并行写入，并会顺序协调每次调用，但业务上仍应确认多路数据的顺序和分帧规则。

为什么 io.Copy 一直不返回？

通常是生产者没有关闭写端，或生产者阻塞且无法继续写入。先检查两端是否并发运行，以及所有返回路径是否执行关闭。