Go io.PipeReader.CloseWithError 怎样把生产者错误传给消费者

我在排查 Go 流式处理的错误丢失时，最先检查的不是 goroutine，而是 CloseWithError 调用在了哪一端。这里很容易写反：生产者要把错误交给消费者，应调用 PipeWriter.CloseWithError(err)；PipeReader.CloseWithError(err) 的作用是读端主动关闭，让后续写入返回这个错误。

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

记住两个方向：PipeWriter.CloseWithError 让消费者的 Read 收到生产者错误；PipeReader.CloseWithError 让生产者的 Write 收到读端关闭原因。传入 nil 时，写端关闭对应读取端的 EOF。

生产目标：先把错误传播方向固定下来

io.Pipe 是一个同步的内存管道，没有内部缓冲。生产者通常持有 *io.PipeWriter，消费者持有 *io.PipeReader。因此错误应该跟着“谁结束了自己的半边”传播：

所以，标题里提到的 PipeReader.CloseWithError 并不能把生产者错误“转发给消费者”。它解决的是反方向的取消和退出；生产端异常应在写端结束时携带出去。

环境准备：io.Pipe 的读写是同步配对的

调用 io.Pipe() 后，每次写入都要等一个或多个读取把数据取走，写端不会像带缓冲 channel 那样悄悄积压数据。这个特性很适合把生产错误直接交给消费循环，但也意味着消费方必须持续读取，或者在取消时关闭读端。

写端传入非 nil 错误后，读端不应该把它改写成普通的结束状态。消费循环要先处理本次读取到的字节，再检查错误，这样不会漏掉“本次同时拿到数据和错误”的接口约定。

安全配置：生产者用 PipeWriter.CloseWithError 结束

下面的最小示例先写入一段数据，再模拟生产异常。生产者调用的是 PipeWriter.CloseWithError，消费者通过 errors.Is 识别原始错误：

package main

import (
	"errors"
	"fmt"
	"io"
)

var errProducer = errors.New("生产者读取上游数据失败")

func main() {
	reader, writer := io.Pipe()
	producerDone := make(chan struct{})

	go func() {
		defer close(producerDone)
		// 同步 Pipe 的 Write 要等消费者 Read，不能把它当成带缓冲队列。
		if _, err := writer.Write([]byte("第一段数据")); err != nil {
			return
		}
		// 生产者异常从写端关闭，并把原始错误交给后续 Read。
		_ = writer.CloseWithError(errProducer)
	}()

	buf := make([]byte, 32)
	for {
		n, err := reader.Read(buf)
		if n > 0 {
			// 即使本次 Read 同时带回错误，也先处理已经收到的字节。
			fmt.Printf("收到数据: %s\n", buf[:n])
		}
		if err == nil {
			continue
		}
		if errors.Is(err, errProducer) {
			// errors.Is 保留了生产者错误的身份，便于上层分类处理。
			fmt.Println("识别到生产者错误:", err)
			break
		}
		if errors.Is(err, io.EOF) {
			// 只有生产者正常关闭，消费者才把它当作自然结束。
			fmt.Println("流正常结束")
			break
		}
		fmt.Println("读取失败:", err)
		break
	}

	// 示例结束前等待生产 goroutine 收尾，避免主函数提前退出。
	

这里的关键不是打印了哪一条日志，而是错误身份没有丢失。若上层需要进一步包装错误，也应使用 %w，让调用方仍能通过 errors.Is 或 errors.As 判断来源。

日志审计：消费者要区分自定义错误和 EOF

PipeWriter.CloseWithError(nil) 等价于正常结束，消费者看到的是 io.EOF。传入业务错误时，读取端拿到的是这个错误本身；因此不要写成“只要 err != nil 就结束并记录为 EOF”，否则线上日志会把生产失败伪装成正常收尾。

如果生产者已经写入了最后一段数据，再调用 CloseWithError，消费者可能先拿到那段数据，下一次读取才收到关闭错误。消费代码应同时检查 n 和 err，不能因为 err 非空就跳过 buf[:n]。

权限边界：PipeReader.CloseWithError 适合读端主动取消

当消费者达到大小上限、命中筛选条件，或者上层请求已经取消时，读端可以调用 PipeReader.CloseWithError。这会关闭读端；生产者后续写入会得到该错误，从阻塞的 Write 中退出。

package main

import (
	"errors"
	"fmt"
	"io"
)

var errConsumerStop = errors.New("消费者已停止读取")

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

	go func() {
		// 没有对应的 Read 时，Write 会阻塞，直到读端关闭或继续消费。
		_, err := writer.Write([]byte("等待消费者处理"))
		writeResult 

这条路径和生产者报错路径不要混用：生产者失败时由 PipeWriter 关闭，消费者取消时由 PipeReader 关闭。两边都可以唤醒对端，但错误的归属不同，日志字段也应分别记录 producer_error 和 consumer_cancel。

发布检查：关闭顺序和首个错误生效

官方实现会保存关闭错误，并且不会用后续调用覆盖已经保存的错误；两个 CloseWithError 方法本身都返回 nil。这意味着调用方不能靠关闭方法的返回值判断对端是否收到了错误，而应在另一端检查 Read 或 Write 的返回值。

1. 生产者每次写入都检查 Write 返回值，收到读端取消后立刻停止后续写入。
2. 生产者完成时调用 writer.Close()，失败时调用 writer.CloseWithError(err)，不要先用 Close() 把错误状态吃掉。
3. 消费者循环同时处理 n 和 err，对 io.EOF 与业务错误分开记录。
4. 消费者提前退出时调用 reader.CloseWithError(err)，避免生产 goroutine 永久卡在 Write。
5. 不要依赖重复关闭来“刷新”错误；首个关闭原因生效，后续关闭不会覆盖它。

最终可以用一句话复盘这类问题：错误从哪一端产生，就由哪一端的关闭方法把原因送到另一端；生产者到消费者用 PipeWriter.CloseWithError，消费者到生产者用 PipeReader.CloseWithError。

相关问题

PipeWriter.CloseWithError(nil) 和 Close 有区别吗？

对消费者看到的结束状态来说没有实质区别，后续读取会返回 io.EOF。如果有真实生产错误，不要传 nil，否则上层只能看到正常结束。

为什么 CloseWithError 的返回值不是传入的错误？

它表示“关闭动作已被接受”，错误是给管道另一端的读写操作使用的。要判断传播结果，应检查对端的 Read 或 Write。

读端关闭后还要继续读吗？

不需要。读端主动关闭后，生产者应停止写入；如果业务还需要把错误交给更上层，应在生产 goroutine 或请求处理层保存原始错误。