Go io.Pipe 消费端退出后生产端阻塞怎么办

io.Pipe 的消费端不再读取后，生产端仍阻塞在 Write，通常需要由消费端调用 pr.CloseWithError(consumeErr)，让等待中的写入返回错误。关闭读端之后再等待生产协程；如果先等生产完成，而生产又在等读端继续消费，就会形成相互等待。

这不是给管道增加缓冲的问题。CloseWithError 把结束原因传给另一端，生产代码仍必须检查 Write 的错误并返回。Go 官方接口文档：https://pkg.go.dev/io#PipeReader.CloseWithError。

先看谁还在等待读取

设想一段记录生成代码写入 PipeWriter，消费代码读完前缀后发现业务字段不合要求，直接返回。生产者可能还在同一次 Write 中等待剩余字节被读取；消费函数返回，本身不会关闭 PipeReader。

io.Pipe 是同步内存管道，没有内部数据缓冲来接收无人读取的剩余内容。一次较大的写入可以由多次读取消耗。因此“已经读到一点内容”与“生产端这次 Write 已完成”不是同一件事。读端决定不再读取，就必须承担结束自己端点的职责。

把两个端点的关闭职责分开

不要因为手上有 pw 就把消费错误只写到写端。题目里的退出发生在读侧，读侧应明确停止对写入的接收。关闭操作的返回值也不是业务失败的回执，应用应保留原来的消费错误和生产结果。

让消费错误结束阻塞的Write

以下完整示例只使用固定的有限数据。consume 有意在读满三个字节后拒绝记录，produce 的一次写入则长于三个字节，从而展示“消费提前退出但写入尚未完成”的场景。代码未在这里运行，输出判断是读者运行时应观察的条件。

package main

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

var errRejected = errors.New("消费端拒绝当前记录")

// produce 检查每次写入结果，不吞掉对端关闭错误。
func produce(w io.Writer) error {
    _, err := w.Write([]byte("record-body-for-demo"))
    if err != nil {
        return fmt.Errorf("生产端写入: %w", err)
    }
    return nil
}

// consume 读取前缀后模拟业务拒绝，不再读取剩余内容。
func consume(r io.Reader) error {
    prefix := make([]byte, 3)
    if _, err := io.ReadFull(r, prefix); err != nil {
        return fmt.Errorf("读取前缀: %w", err)
    }
    return errRejected
}

func main() {
    pr, pw := io.Pipe()
    defer pr.Close()
    done := make(chan error, 1)

    go func() {
        err := produce(pw)
        // 写端完成时把生产结果交给读端；nil 表示正常结束。
        _ = pw.CloseWithError(err)
        done 

这里没有靠 Sleep 猜测生产者是否已经写入。消费端读满前缀后关闭 pr，尚未完成的写入才能返回；生产者检查错误、关闭自己的 pw，并通过 done 交回结果。缓冲为一的 done 用于结果交接，不是管道数据缓冲。

如果把 produceErr := 移到读端 CloseWithError 之前，这个例子的消费者不再读，生产者却还在写，主协程便无法走到关闭语句。把 defer pr.Close 放在函数尾部也不足以解决函数内部先等待的情况，因为 defer 要等函数返回才执行。

观察返回原因，而不是只看没有卡住

对于本例，应观察消费端保留 errRejected，生产端返回的包装错误也能通过 errors.Is 匹配它，并且 done 被取回。这些条件共同说明消费原因传到了写侧，且生产协程完成了收尾；它们不表示记录导入成功。

实际应用不要只忽略 produceErr 后返回 nil。消费端拒绝记录就是业务失败，即使生产者因相同原因退出，也应保留这个失败结果。若两侧出现独立错误，应用可以分别记录或按自己的优先级汇总，不能让关闭操作返回 nil 掩盖真实错误。

CloseWithError 不是任意任务取消器

这个办法解决的是管道读写的等待。如果生产者卡在另一个数据源的 Read、数据库查询或外部调用，关闭 PipeReader 不会自动取消那些操作。上游任务还需自己的取消或关闭约定，不能因使用了 io.Pipe 就省略其他资源收尾。

消费端“只要前缀就算成功”也要关闭读端，但 nil 关闭不会向写端表达业务成功；写端会看到 ErrClosedPipe。若希望把预期早停识别为正常业务结果，可以定义应用自己的早停原因并单独分类，不能把任意 ErrClosedPipe 都无条件忽略。

还要避免抢先关闭丢失原因：若先 defer 或其他路径关闭某端，再试图 CloseWithError 写入更具体错误，先前记录的关闭原因不会被后来的错误替换。设计上应让端点拥有者在原因明确时关闭，并让生产函数始终检查写入结果。

相关关闭问题

读端 CloseWithError(nil) 等同于给写端发送 EOF 吗？

不是。读端 nil 关闭使写端得到 ErrClosedPipe；写端 nil 关闭才对应读端最终读到 EOF。两端的正常结束含义不同。

CloseWithError 自己返回 nil，错误是不是丢了？

它的 nil 返回值不负责携带业务原因。原因由另一端的 Read 或 Write 观察，业务原始错误应由调用方自己保存。

关闭读端后还需要等待生产者吗？

需要在拥有生产协程的层面完成结果回收。关闭管道只是解除相关阻塞，不等于你已经观察到生产函数退出；上面的 done 正是用于这个目的。

两个端点的关闭与读写规则见 https://pkg.go.dev/io#Pipe；实现中的错误保存和唤醒逻辑可参考 https://go.dev/src/io/pipe.go。