Go io.Pipe 的生产者消费者错误传播

用 io.Pipe 把一个生产者接到一个消费者时，最容易误判的是“写入成功”与“业务处理成功”之间的关系。它没有内部缓冲队列：Write 会等待读端把本次数据接走。因此，生产失败应该由写端调用 CloseWithError 传给读端；读端提前放弃时，也应关闭自己的半边，让生产者从阻塞的 Write 中返回。

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

正常结束用 Close 产生 io.EOF；异常结束用对应半边的 CloseWithError 传递真实错误。不要只关闭管道后丢掉错误，也不要让一端退出而另一端继续等待。

io.Pipe 为什么会让生产者等待

io.Pipe() 返回一对同步连接的 *io.PipeReader 和 *io.PipeWriter。写入端的数据要被读端消费后，Write 才能完成；如果读端尚未开始读取，生产者可能停在写入调用上。这种特性适合把编码器、压缩器或分块生成器直接接给上传逻辑，但它不是用来积压突发数据的缓存。

package main

import (
	"fmt"
	"io"
)

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

	go func() {
		// Write 会等待读端消费本次数据后才返回。
		_, err := writer.Write([]byte("order-42"))
		if err != nil {
			fmt.Println("producer:", err)
			return
		}
		// 正常结束由写端关闭，读端随后会收到 io.EOF。
		_ = writer.Close()
	}()

	buf := make([]byte, 32)
	n, err := reader.Read(buf)
	fmt.Printf("data=%q err=%v\n", buf[:n], err)
}

这里的关键不是把数据写进去，而是让两端都拥有明确的结束路径。写端正常关闭后，读端会在剩余数据读完时看到 io.EOF；如果读端提前关闭，写端的 Write 会收到关闭相关错误。

由写端把生产错误传给消费者

生产者通常负责读取文件、生成压缩流或转换记录。只要中途发生错误，就不要直接返回让读端永远等下去，而是调用 writer.CloseWithError(err)。官方文档定义了这个方向：之后的 Read 会返回该错误，而不是把它伪装成 EOF。

func produce(writer *io.PipeWriter) {
	defer writer.Close()

	for i := 0; i 

同一个半边的关闭状态只保留第一次有效的关闭错误，后续关闭不会覆盖已经记录的错误。因此，调用方应尽量在最接近根因的位置传入原始错误；不要先用一个泛化错误关闭，再期待后面补写细节。

读端提前退出时，怎样让生产者停下来

消费者可能只需要前几条记录，或者发现响应已经取消。此时如果只从函数返回而不关闭 PipeReader，生产者下一次 Write 仍可能等待。读端可以用 CloseWithError 把“停止消费”的原因传回写端。

func consumeFirst(reader *io.PipeReader) error {
	defer reader.Close()

	buf := make([]byte, 64)
	if _, err := reader.Read(buf); err != nil {
		return err
	}
	// 业务只需要一块数据，明确通知生产者停止写入。
	stopErr := errors.New("消费者已完成所需数据")
	_ = reader.CloseWithError(stopErr)
	return nil
}

func produceUntilStopped(writer *io.PipeWriter) error {
	defer writer.Close()
	for {
		if _, err := writer.Write(nextChunk()); err != nil {
			// 这里能收到读端的停止原因，及时结束生产循环。
			return fmt.Errorf("producer stopped: %w", err)
		}
	}
}

这条路径的判断重点是：写端拿到的错误代表“读端已经关闭或拒绝继续接收”，不一定是生产数据本身损坏。日志中应把两类错误区分开，避免把正常的消费取消误报成上游故障。

用一个协调函数收口两端生命周期

实际代码里，最稳妥的做法是让一个协调层同时负责启动生产协程、消费数据、关闭读端以及等待生产协程结束。下面的示例把生产错误和读端退出都收敛到一个结果中；代码中的 nextChunk 代表真实业务的分块生成函数。

func pipeJob() error {
	reader, writer := io.Pipe()
	producerDone := make(chan error, 1)

	go func() {
		// 生产协程只负责写入，并把最终结果送回协调层。
		producerDone 

示例的核心是“谁发现问题，谁负责解除等待”：生产者发现上游错误就通过写端传播；消费者发现自己不能继续就关闭读端；协调层等待生产协程回收，避免后台 goroutine 脱离生命周期。

几个容易混淆的边界

另外，io.Pipe 的同步特性意味着写端没有缓冲来吸收速度差。需要排队或解耦时，应在架构层选择带容量的通道或专用队列；不要靠启动更多生产 goroutine 来掩盖没有消费者的问题。

采用前的速查结论

• 把 Close 留给正常结束，把 CloseWithError 留给需要传递的异常或取消原因。
• 读端只消费部分数据时，必须关闭 PipeReader，否则写端可能永久卡在下一次 Write。
• 不要把读到的非 EOF 错误当成普通结束；它通常是生产端显式传来的根因。
• 用 channel 或其他协调方式等待生产协程结束，保证错误不会只停留在后台日志里。

相关问题

io.Pipe 有缓存吗？没有，它把读写直接同步配对，写入不会像带缓冲队列那样立即返回。

为什么读端收到的是 EOF？通常表示写端用普通 Close 正常结束，且剩余数据已经读完。

CloseWithError(nil) 会怎样？对写端来说等价于关闭后让读端得到 EOF；它不适合表达一个真实失败原因。