流式处理发生中途错误时，怎样让读写两端都及时退出

我在排查一条“读取压缩包、边读边解码、再写入对象存储”的 Go 流水线时，遇到过一个很隐蔽的现象：下游已经报错返回了，上游 goroutine 还卡在写管道，连接和请求迟迟不释放。真正可靠的做法不是只调用一次 cancel()，而是把取消原因、管道两端和底层可关闭资源绑成同一个退出协议。

把取消信号传到两端，而不是只停一个循环

这类流通常有三个边界：生产端从 src 读取，io.Pipe 在中间传递字节，消费端把数据写到 dst。任一边先失败，都要让另外两边尽快知道。Context 适合传播请求取消和超时；Pipe 的 CloseWithError 适合把流级错误送到对端。两者职责不同，不能互相替代。

用一个停止函数收拢错误和资源释放

下面的示例把输入限定为 io.ReadCloser，这是刻意的边界：如果底层读取永远阻塞、又没有关闭方法，任何上层取消都无法凭空打断它。示例不依赖第三方包，核心是让所有失败路径都调用同一个 stop。

package main

import (
    "context"
    "errors"
    "io"
    "sync"
)

func relay(ctx context.Context, src io.ReadCloser, dst io.Writer) error {
    ctx, cancel := context.WithCancelCause(ctx)
    defer cancel(nil) // 正常返回时释放派生 Context 的资源

    pr, pw := io.Pipe()
    var stopOnce sync.Once
    stop := func(err error) {
        if err == nil {
            err = context.Canceled
        }
        stopOnce.Do(func() {
            cancel(err)                 // 广播取消原因
            _ = src.Close()             // 唤醒可能阻塞的底层 Read
            _ = pr.CloseWithError(err)  // 唤醒消费端
            _ = pw.CloseWithError(err)  // 唤醒生产端的 Write
        })
    }

    results := make(chan error, 2)
    go func() {
        _, err := io.Copy(pw, src)
        if err != nil {
            stop(err) // 上游读错或管道写错，都通知另一端
        } else {
            _ = pw.Close() // 只有正常 EOF 才发送正常结束
        }
        results 

这里的关键不在 io.Copy 本身，而在四个出口都能抵达 stop：源端读取失败、目标端写失败、外部 Context 取消，以及管道端点被关闭。sync.Once 让多个 goroutine 同时发现错误时只有第一个原因负责关闭资源。

为什么必须同时关闭管道和底层资源

只调用 cancel，只能让主动检查 ctx.Done() 的代码看到信号；已经卡在 PipeWriter.Write 或真实网络读取里的 goroutine，不一定会自动醒来。io.Pipe 没有内部缓冲，写入要等待读端消费；读端关闭后，写端才会得到错误。因此失败路径至少要覆盖三件事：给管道两端注入同一个错误、关闭底层 ReadCloser、等待两端都汇报。

如果真实数据源是 *os.File、http.Response.Body 或自定义网络流，应该把它的关闭动作纳入同一生命周期。若数据源的 Read 不响应 Close，就只能在数据源层增加可取消 API；不要用定时器掩盖 goroutine 泄漏。

上线前检查四个退出边界

我通常会给这条流水线留一张很短的复查清单：源端故意返回错误时，目标端是否退出；目标端写入失败时，源端是否不再生产；请求取消后，是否能看到底层连接关闭；正常 EOF 是否仍被当作成功。日志里同时记录“第一个错误”和“收尾错误”，就能区分根因与 ErrClosedPipe 这类连带信号。

相关问题

只用一个 done channel 可以吗？

可以表达取消，但它不会自动关闭网络连接或传递流级错误。对于跨函数、跨请求的链路，Context 加管道端点关闭更容易统一资源边界。

为什么不直接忽略第二个 goroutine 的错误？

第二个错误可能暴露真实的资源未关闭问题。应保留第一个根因，同时等待另一个 goroutine 退出；不要为了“看起来成功”提前返回。

低于 Go 1.20 怎么保留错误原因？

可以用 context.WithCancel 配合一个受保护的错误变量，或显式传递错误通道。无论选哪种写法，关闭 Pipe 两端和底层 ReadCloser 的原则不变。