Go io.Pipe 怎样让消费者感知生产端错误

用 Go 把一个生产端接到一个消费端时，io.Pipe 很适合做“边写边读”的同步桥接，但它有一个容易被忽略的边界：生产端如果直接返回错误，消费者并不会自动知道这个错误是什么。真正稳妥的做法是让生产端在退出前调用 PipeWriter.CloseWithError(err)，再让消费者保留 PipeReader.Read 或 io.ReadAll 返回的错误。

官方文档：https://pkg.go.dev/io#Pipe

核心配方是：生产端负责把业务错误写入 PipeWriter，消费端负责读取并分类；正常结束用 EOF，异常结束用 CloseWithError 传递原始错误。

先看清 Pipe 两端的同步边界

io.Pipe() 返回一对 *io.PipeReader 和 *io.PipeWriter。它不是一个默认无限增长的缓冲队列：写入和读取需要在两端配合，生产端写得太快时可能等待消费端，消费端读不到数据时也会等待生产端。这个特性正适合把编码器、压缩器或流式生成器接到下游处理器。

先把两个端点的责任写清楚：生产端只负责写数据和报告生产错误，消费端只负责读数据和解释读到的终止错误。不要让消费端猜测生产端是否已经失败。

package stream

import (
	"fmt"
	"io"
	"sync"
)

// CopyFromProducer 将生产函数产生的数据交给消费函数，并保留生产错误。
func CopyFromProducer(produce func(io.Writer) error, consume func(io.Reader) error) error {
	reader, writer := io.Pipe()
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		// 生产端结束时必须关闭写端，否则消费端可能一直等待 EOF。
		err := produce(writer)
		if err != nil {
			_ = writer.CloseWithError(err)
			return
		}
		_ = writer.Close()
	}()

	err := consume(reader)
	// 消费函数返回后关闭读端，及时释放仍可能阻塞的写端。
	_ = reader.Close()
	wg.Wait()
	if err != nil {
		return fmt.Errorf("消费端失败: %w", err)
	}
	return nil
}

这个骨架的关键不在包装函数本身，而在生产端的收尾分支：成功时关闭得到 EOF，失败时用 CloseWithError 保存原因。若遗漏关闭，读端的结束条件就不完整。

用 CloseWithError 把生产错误传给读端

生产端错误不应该只写日志然后返回。对 io.Pipe 来说，写端的 CloseWithError 会把指定错误交给另一端的读取操作；消费端因此可以继续沿用标准的 errors.Is 或 errors.As 做判断。

package stream

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

var errBadRecord = errors.New("记录格式不完整")

// ReadAllRecords 读取整个流，并把生产端传来的错误保留给调用方。
func ReadAllRecords(reader *io.PipeReader) ([]byte, error) {
	data, err := io.ReadAll(reader)
	if err == nil {
		return data, nil
	}
	// errors.Is 可以识别被 CloseWithError 包装后仍保留的业务错误。
	if errors.Is(err, errBadRecord) {
		return nil, fmt.Errorf("生产数据无效: %w", err)
	}
	// EOF 表示生产端正常关闭，不应和异常混为一谈。
	if errors.Is(err, io.EOF) {
		return data, nil
	}
	return nil, fmt.Errorf("读取流失败: %w", err)
}

示例中 io.ReadAll 把读取过程收拢成一次结果；如果下游需要持续处理大数据，则应改为循环调用 Read，分类逻辑保持不变。重要的是不要丢掉原始错误，否则上层只能看到“读取失败”，无法判断是生产数据错误、读端提前关闭还是其他 I/O 问题。

把 EOF、业务错误和提前关闭分开

正常结束、生产失败和消费端主动停止是三种不同状态。成功的生产端调用 Close()，读端通常以 io.EOF 结束；生产端调用 CloseWithError(err)，读端拿到的是该错误；消费端如果已经不需要数据，应该关闭读端，让生产端后续写入尽快返回错误。

package stream

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

// ClassifyReadError 将读端结果转成业务层可处理的三类状态。
func ClassifyReadError(err error) string {
	switch {
	case err == nil:
		return "继续处理数据"
	case errors.Is(err, io.EOF):
		return "生产端正常结束"
	case errors.Is(err, errBadRecord):
		return "生产端报告业务错误"
	case errors.Is(err, io.ErrClosedPipe):
		return "某一端已经提前关闭"
	default:
		return fmt.Sprintf("其他读取错误: %v", err)
	}
}

不要用错误字符串做判断。CloseWithError 的意义就是把错误值留在错误链里，调用方可以使用 errors.Is；如果业务错误还带有字段，则使用 errors.As 取回具体类型。

用单一收尾路径避免 goroutine 泄漏

一个常见泄漏点是：消费端提前返回，但生产 goroutine 仍在 Write 上等待。解决方法是消费端返回前关闭读端，并由生产端统一负责关闭写端；同时用 WaitGroup 等待生产 goroutine 完成，这样调用函数返回时资源边界是完整的。

package stream

import (
	"fmt"
	"io"
	"sync"
)

// StreamWithEarlyStop 演示消费端停止后如何让生产端退出。
func StreamWithEarlyStop(produce func(io.Writer) error, consume func(io.Reader) error) error {
	reader, writer := io.Pipe()
	var wg sync.WaitGroup
	wg.Add(1)
	var producerErr error
	go func() {
		defer wg.Done()
		producerErr = produce(writer)
		if producerErr != nil {
			_ = writer.CloseWithError(producerErr)
			return
		}
		_ = writer.Close()
	}()

	consumerErr := consume(reader)
	// 无论消费端成功还是提前停止，都先关闭读端解除写端阻塞。
	_ = reader.Close()
	wg.Wait()
	if consumerErr != nil {
		return fmt.Errorf("消费端提前停止: %w", consumerErr)
	}
	return producerErr
}

真实项目还应根据并发模型决定是否需要额外的取消信号；这里的边界只覆盖 io.Pipe 两端的关闭和错误传播。不要让多个 goroutine 无序地重复关闭同一个端点，也不要在没有等待生产 goroutine 的情况下直接返回。

几个容易混淆的问题

生产端返回 error，读端一定能拿到它吗？

不一定。只有生产端把错误传给 CloseWithError，读端的后续读取才有机会得到这个错误；只从生产函数返回而不关闭写端，可能让读端继续等待。

什么时候应该使用 Close，而不是 CloseWithError？

生产端已经正常写完所有数据时使用 Close，让读端以 EOF 结束；只有需要说明异常原因时才使用 CloseWithError。

消费端提前退出后为什么还要关闭 reader？

因为写端可能正阻塞在下一次写入。关闭读端会把“下游不再消费”传回写端，生产 goroutine 才有机会结束，调用方也能在等待后安全返回。

归纳起来，io.Pipe 的稳定用法是把“谁负责关闭”和“错误从哪一端传来”先约定清楚：成功路径关闭写端得到 EOF，异常路径使用 CloseWithError，消费端提前停止则关闭读端并等待生产 goroutine。