Go io.Pipe 怎么把生产端错误传给读取端

在流式导出、压缩或转换任务里，生产端可能已经写入一部分数据，随后才发现源文件损坏或上游返回错误。此时如果直接调用 PipeWriter.Close()，读取端通常只会得到 io.EOF，很容易把半截结果当成完整结果。正确做法是：生产逻辑失败时调用 CloseWithError(err)，读取端先处理已经读到的字节，再判断返回的错误。

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

io.Pipe 不会自动替生产端传递业务错误。PipeWriter.CloseWithError(err) 会让读取端在管道结束时拿到这个错误；传入 nil 则表示正常结束，读取端得到 io.EOF。

先把 io.Pipe 的同步边界看清楚

io.Pipe 是一个同步的内存管道，没有内部缓冲。写入端的 Write 会等待读取端消费数据；读取端的 Read 会等待写入端提供数据或关闭管道。它适合把“生成数据”的代码接到“消费数据”的代码，但不会替你设计错误协议。

因此，生产端的函数返回值和读取端的返回值是两条不同的错误观察点：

生产端失败时用 CloseWithError 传递原始错误

生产逻辑完成后调用 Close()，读取端会把结束理解为正常 EOF；生产逻辑失败后调用 CloseWithError(err)，读取端才能知道这是异常结束。这个错误不需要先转成字符串，保留原始值可以让消费端继续使用 errors.Is 或 errors.As。

package main

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

var errSourceBroken = errors.New("source stream is broken")

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

	go func() {
		// 只有正常返回时才用 Close 表示 EOF；失败分支会先传递原始错误。
		defer writer.Close()

		if _, err := io.Copy(writer, strings.NewReader("part-1\n")); err != nil {
			// 读取端提前关闭时，写入错误应直接结束生产协程。
			return
		}

		// 模拟生产端在写入部分数据后发现上游失败。
		_ = writer.CloseWithError(errSourceBroken)
	}()

	data, err := io.ReadAll(reader)
	fmt.Printf("data=%q\n", data)
	fmt.Printf("same error=%t\n", errors.Is(err, errSourceBroken))
}

这个例子里，读取端可能已经拿到了 part-1，但 io.ReadAll 最终仍会返回生产端传入的错误。defer writer.Close() 只是兜底关闭；官方实现不会用后一次关闭覆盖已经保存的首个错误。

读取端要先处理字节，再处理错误

Go 的 Reader 约定允许一次 Read 同时返回 n > 0 和非空错误。对管道来说，这一点尤其重要：生产端可能已经发送了最后一段数据，然后才关闭并携带错误。消费端不能因为看到错误就丢掉这段数据。

package main

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

func consume(reader *io.PipeReader, producerErr error) error {
	buf := make([]byte, 32)
	for {
		n, err := reader.Read(buf)
		if n > 0 {
			// 先落盘、解码或交给下游，避免丢失错误前已经到达的数据。
			fmt.Printf("chunk=%q\n", buf[:n])
		}
		if err == nil {
			continue
		}

		if errors.Is(err, io.EOF) {
			// EOF 表示生产端正常关闭，当前结果可以按成功收尾。
			return nil
		}
		if errors.Is(err, producerErr) {
			// 保留原始错误语义，调用方可以决定重试或丢弃半成品。
			return fmt.Errorf("consume pipe: %w", err)
		}
		return err
	}
}

生产端如果写入成功后再调用 CloseWithError，读取端会先拿到已到达的字节，随后在某次读取中看到该错误。调用方应明确“已经消费的部分是否可用”，不要把 n > 0 的数据静默丢掉。

Close、CloseWithError 和 ErrClosedPipe 有什么区别

可以把关闭方向分成三种情况。写入端关闭影响读取端，读取端关闭影响后续写入；两者都调用 CloseWithError 时，传入的非空错误只对相反方向可见。

CloseWithError 本身总是返回 nil，它的作用是记录关闭原因，而不是把原因作为这个关闭调用的返回值交给调用者。因此不要写成“检查 CloseWithError 的返回值来判断生产失败”；生产失败的判断应由读取端通过 Read 返回值，或由生产协程自己记录。

生产代码里的关闭约定怎么写

实际项目中可以把关闭责任固定成一条小规则：生产函数成功时只调用 Close，失败时只调用一次 CloseWithError；读取端提前放弃时调用 reader.Close，生产端检查后续 Write 的错误并尽快返回。

func produce(writer *io.PipeWriter) {
	if err := generate(writer); err != nil {
		// 把生成阶段的错误交给读取端，不要把它降级成 EOF。
		_ = writer.CloseWithError(err)
		return
	}

	// 没有错误才使用 EOF 作为正常结束信号。
	_ = writer.Close()
}

如果生产端还需要记录日志，可以在调用 CloseWithError 前记录任务标识、已写字节数和错误类型；读取端则记录自己是否已经消费了部分内容。这样能区分“源数据失败”“读取端主动放弃”和“真正的正常结束”。

结论与常见问题

解决 Go io.Pipe 生产端错误传递的关键只有一条：正常完成调用 Close，异常结束调用 CloseWithError(err)。读取端先处理 n > 0 的数据，再用 errors.Is 判断是否为生产错误；不要把所有非空错误都当作 EOF，也不要只检查 CloseWithError 的返回值。

CloseWithError(nil) 会传递什么？

它等价于正常关闭，读取端在没有更多数据后得到 io.EOF。需要表达失败时必须传入非空错误。

读取端为什么可能先拿到部分数据？

关闭只表示生产端不会再提供数据，已经写入并被管道交付的字节仍然应该先被消费。因此读取循环要先判断 n > 0，再处理 err。

读取端提前关闭后，生产端该检查什么？

检查后续 Write 的错误。读取端用普通 Close 关闭后，写入端通常会看到 io.ErrClosedPipe；如果读取端用 CloseWithError，写入端可看到传入的错误。

io.Pipe 适合做带缓冲的生产消费队列吗？

不适合把它当作缓冲队列。io.Pipe 没有内部缓冲，写入和读取是同步配对的；如果需要解耦生产速度与消费速度，应另选有明确容量和关闭协议的队列结构。