Go Pipe 怎样把生产者错误传给消费者

Go 的 io.Pipe 是同步的内存管道：生产者写入 PipeWriter，消费者从 PipeReader 读取。生产者如果只是调用 Close()，消费者最终只能看到正常的 io.EOF；如果希望把真正的生产错误传给消费者，应在失败分支调用 CloseWithError(err)。

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

最小规则是：生产成功时用 writer.Close() 结束并让读取端得到 EOF；生产失败时用 writer.CloseWithError(err) 结束，读取端会在后续 Read 中收到这个错误。消费者要先处理已经读到的字节，再判断错误。

先分清 Pipe 两端的错误方向

io.Pipe() 返回一对同步端点。写端的关闭结果会影响读端：写端普通关闭代表没有更多数据，读端收到 io.EOF；写端带错误关闭则把这个错误作为读端后续读取的错误。管道没有内部缓冲，写入通常要等读端消费，因此生产者和消费者应当由不同的 goroutine 或不同的执行路径配合。

reader, writer := io.Pipe()

// 生产者只拥有写端，消费者只拥有读端。
go produce(writer)
consume(reader)

错误方向不要反过来理解：PipeWriter.CloseWithError 影响的是读取端后续的 Read；读端关闭后，生产者在写入时才可能看到读端的关闭错误。本文只讨论“生产者失败怎样通知消费者”这条路径。

让生产者在成功和失败分支分别关闭

生产函数的关键是把“正常结束”和“异常结束”写成两条明确路径。下面的示例模拟分段读取数据，其中第二段故意产生业务错误。生产者在错误分支调用 CloseWithError，这样消费者不会把不完整的数据误判成正常结束。

package main

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

var errSourceUnavailable = errors.New("上游数据源不可用")

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

	chunks := []string{"第一段数据\n", "第二段数据\n"}
	for i, chunk := range chunks {
		if i == 1 {
			// 失败必须携带原始原因，不能用普通 Close 把错误伪装成 EOF。
			_ = w.CloseWithError(errSourceUnavailable)
			return
		}
		if _, err := io.WriteString(w, chunk); err != nil {
			// 读端提前退出时，写入错误已经是生产者应返回的结果。
			return
		}
	}
	// defer 中的 Close 只负责正常收尾；错误路径已经记录了错误。
}

func consume(r *io.PipeReader) error {
	buf := make([]byte, 8)
	for {
		n, err := r.Read(buf)
		if n > 0 {
			// 即使本次 Read 同时返回数据和错误，也要先处理有效字节。
			fmt.Printf("收到：%s", buf[:n])
		}
		if err != nil {
			if errors.Is(err, io.EOF) {
				// EOF 表示生产者正常结束，不是业务失败。
				return nil
			}
			return err
		}
	}
}

func main() {
	r, w := io.Pipe()
	go produce(w)

	if err := consume(r); err != nil {
		fmt.Println("消费失败：", err)
	}
}

这里的 defer w.Close() 不会覆盖先前的错误：官方文档规定 CloseWithError 不会被后续关闭改写。实际项目中也可以不依赖这个细节，改成每个分支只调用一次关闭方法，让生命周期更直观。

消费者同时处理数据和最终错误

Reader.Read 的约定允许一次调用同时返回已经读到的字节和一个错误。因此消费者不能写成“只要 err != nil 就直接返回”，否则可能丢掉最后一小段有效数据。正确顺序是先处理 n > 0，再判断 err。

func copyAndReport(r *io.PipeReader, dst io.Writer) error {
	buf := make([]byte, 32*1024)
	for {
		n, err := r.Read(buf)
		if n > 0 {
			// 先转发本次已读数据，避免错误和数据同时返回时丢尾部内容。
			if _, writeErr := dst.Write(buf[:n]); writeErr != nil {
				return fmt.Errorf("写入目标失败: %w", writeErr)
			}
		}
		if err != nil {
			// EOF 是正常完成，其它错误都要保留给上层处理。
			if errors.Is(err, io.EOF) {
				return nil
			}
			return fmt.Errorf("读取生产流失败: %w", err)
		}
	}
}

如果上游用 CloseWithError(err) 结束，消费者拿到的是原始错误或其包装形式，可以用 errors.Is 判断哨兵错误。不要只根据已收到的部分数据猜测成功，因为部分数据后面仍可能跟着生产失败。

把错误传播封装成可复用函数

把生产逻辑放进一个返回 error 的函数，外层 goroutine 负责把返回值转成管道状态，会比在多个业务分支里手写关闭更容易维护。关键是：业务函数返回 nil 才普通关闭，返回非空错误才调用 CloseWithError。

func runProducer(w *io.PipeWriter) {
	// 业务函数只描述生产结果，关闭策略集中在这一层。
	err := produceData(w)
	if err != nil {
		// 让消费者在 Read 端收到可追踪的业务原因。
		_ = w.CloseWithError(err)
		return
	}
	// nil 表示正常结束，消费者应看到 EOF。
	_ = w.Close()
}

func produceData(w io.Writer) error {
	if _, err := io.WriteString(w, "可消费的数据\n"); err != nil {
		// 保留底层写入错误，让桥接层决定如何传播。
		return fmt.Errorf("生成数据失败: %w", err)
	}
	return errors.New("示例业务错误")
}

这个结构也适合从文件、压缩流或网络响应中逐段生成数据。生产函数不需要知道消费者如何处理错误，只需要遵守返回值契约；消费者则把读取错误带回调用方或取消上层任务。

排查 CloseWithError 不生效的边界

把错误写在了读端的 Close 之后

如果消费者已经调用 r.Close()，生产者后续写入可能收到 io.ErrClosedPipe。这表示消费方主动放弃，不应再期待原生产错误能正常传到读端。

生产者使用了普通 Close

w.Close() 的语义是正常结束，读端会看到 EOF。只要错误仍然需要向上游报告，就必须在错误发生的分支调用 w.CloseWithError(err)，而不是先 Close 再补报错误。

重复关闭或错误顺序混乱

CloseWithError 不会覆盖已经记录的错误。不要让多个 goroutine 同时决定管道的最终错误；最好由一个生产者拥有写端，并由一个明确的出口负责关闭。

没有持续读取导致写端阻塞

io.Pipe 没有内部缓冲，写端会等待读端消费。如果消费者在收到某个错误后提前返回，生产者可能卡在下一次写入。消费者放弃时要关闭读端，生产者检查写入错误并退出，避免 goroutine 泄漏。

小结

Go 用 io.Pipe 传递生产错误时，记住三件事：生产成功调用 PipeWriter.Close 让读取端得到 EOF；生产失败调用 PipeWriter.CloseWithError(err) 让读取端收到原因；消费者每次读取都先处理 n > 0 的数据，再区分 EOF 和业务错误。把关闭策略集中到一个生产桥接层，就能避免普通关闭、重复关闭和提前退出造成的误判。

相关问题

• io.Pipe 会不会缓存大量数据？不会，它是同步管道，写入与读取需要配合，不适合用来替代带容量的缓冲队列。
• 为什么读端拿到的是 EOF 而不是业务错误？通常是生产者走了普通 Close，或者错误发生在关闭之后；检查错误分支是否直接调用了 CloseWithError。
• 消费者提前退出时生产者怎么办？关闭读端，让生产者的下一次写入返回关闭错误，并让生产 goroutine 及时结束。