Go io.Pipe 串流生产者消费者的背压处理

我第一次把大对象从生成端直接送到上传端时，最容易误判的是：io.Pipe 看起来像一个队列，实际上更像一条没有仓库的传送带。它没有内部缓冲，PipeWriter.Write 会等到 PipeReader.Read 把这次写入的数据消费完，因此消费者变慢时，生产者会自然停下来，这就是串流场景里的背压。

这篇只处理四件事：同步读写、正常关闭、错误传播和取消。官方文档把 io.Pipe 定义为同步内存管道，适合把需要 io.Reader 的代码和需要 io.Writer 的代码直接连接起来；它不是带容量的消息队列。

官方参考：https://pkg.go.dev/io#Pipe

流水线取消参考：https://go.dev/blog/pipelines

先把同步管道当成天然背压

最小结构是一个 goroutine 持有写端，当前 goroutine 读写端。写端没有可供自己无限堆积的数据区，所以消费者没有及时读取时，生产者就会停在 Write。这能限制内存增长，但也意味着读端必须持续推进。

package main

import (
	"fmt"
	"io"
)

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

	go func() {
		defer writer.Close()
		// 写端按块提供数据；每次 Write 都要等读端消费。
		for _, chunk := range []string{"header\n", "body\n", "tail\n"} {
			if _, err := io.WriteString(writer, chunk); err != nil {
				// 读端提前关闭时，停止继续生产。
				fmt.Println("write failed:", err)
				return
			}
		}
	}()

	// Reader 端可以交给任意需要 io.Reader 的下游。
	if _, err := io.Copy(io.Discard, reader); err != nil {
		fmt.Println("read failed:", err)
	}
}

上面的重点不是把字符串写进管道，而是让生产和消费的速度互相可见。和带缓冲的 channel 不同，io.Pipe 不会因为先写几块就继续积压；它把“下游目前能吃多少”直接反馈给写端。

正常结束交给 PipeWriter.Close

生产端确定不会再写入时，应关闭 PipeWriter。读端继续读取已经交付的数据，写端正常关闭后，后续读取以 io.EOF 结束。最常见的责任划分是：创建并写入的人负责关闭写端，消费的人只读取并处理 EOF。

func streamText(writer *io.PipeWriter, parts []string) {
	defer writer.Close()
	for _, part := range parts {
		// 写完全部片段后由同一个 owner 关闭，读端才能收到 EOF。
		if _, err := io.WriteString(writer, part); err != nil {
			// 读端已关闭时，结束生产并保留 Write 返回的错误。
			return
		}
	}
}

不要把 Close 当成“刷新缓冲区”的替代品；io.Pipe 没有内部缓冲。这里的关闭动作表达的是数据终点，让阻塞中的读端知道不会再有新数据。

用 CloseWithError 传递生产失败

如果生产端不是正常结束，而是读取源文件、编码或上游接口失败，直接调用 Close() 会把真实原因丢掉。此时用 PipeWriter.CloseWithError(err) 收口，读端后续的 Read 会得到这个错误，而不是普通的 EOF。

func forward(src io.Reader, writer *io.PipeWriter) {
	// Copy 负责把上游错误带回生产端，再传递给读端。
	if _, err := io.Copy(writer, src); err != nil {
		_ = writer.CloseWithError(err)
		return
	}
	// nil 表示正常结束，读端最终看到 EOF。
	_ = writer.CloseWithError(nil)
}

消费者可以按三类结果处理：err == nil 继续读；err == io.EOF 表示正常结束；其他错误代表生产失败或读端关闭。不要只打印“读取结束”，否则线上很难区分完整上传和中途失败。

读端提前退出时主动解除阻塞

背压的另一面是：如果消费者只拿到第一段数据就返回，生产者可能永远卡在下一次 Write。io.Pipe 没有直接接收 context.Context 的参数，所以取消路径要主动关闭读端，让阻塞写入返回。

func copyWithCancel(ctx context.Context, src io.Reader, dst io.Writer) error {
	reader, writer := io.Pipe()
	stopCancel := make(chan struct{})
	cancelWatcherDone := make(chan struct{})

	go func() {
		defer close(cancelWatcherDone)
		select {
		case 

这个结构里有三条边界：生产端通过关闭写端表达 EOF 或生产错误，消费端通过关闭读端表达“我不再接收”，context 只负责触发取消。若读端主动关闭，写端通常会收到 io.ErrClosedPipe 或读端传入的错误；这时应停止继续生产，而不是无条件重试。

按场景选择 Pipe，而不是把它当缓冲队列

我的判断是：当两个组件本来就通过 io.Reader/io.Writer 对接，而且不希望把完整内容放进内存，io.Pipe 很顺手；当需求是排队、缓冲、广播或重试，它就不是替代 channel、文件或消息系统的通用答案。

几个容易混淆的问题

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

没有内部缓冲。写入和读取需要配对，Write 会等到一个或多个 Read 把本次写入的数据消费完，所以它的背压是同步发生的。

读端关闭后，写端应该继续重试吗？

通常不应该。读端关闭意味着下游已经放弃，写端应识别 io.ErrClosedPipe 或对应错误并退出；只有业务明确允许重新建立一条新管道时才考虑重试。

为什么不用 context 直接传给 io.Pipe？

io.Pipe 的接口只负责 Reader/Writer 配对，不负责取消。需要取消时，把 context 作为旁路信号，在取消分支关闭相应端点，让阻塞的读写调用获得返回机会。