用 io.Pipe 边生成边上传数据而不落整包临时文件

可以。io.Pipe 能把“只会向 io.Writer 写数据”的生成器，直接连接到“需要 io.Reader 作为请求体”的上传客户端。生成一段，HTTP 客户端就读取并发送一段，不需要先把完整 ZIP 写进临时文件，也不必把整包放进内存。

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

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

我为什么把临时 ZIP 换成 io.Pipe

我第一次处理这类任务时，流程很直观：先把几百万条记录写成 ZIP 临时文件，再打开文件上传。它能工作，但包越大，问题越明显：必须等整包生成完才能开始发送，需要预留双份磁盘空间，失败后还要清理半成品，容器的临时盘也可能先被写满。

换成 io.Pipe 后，生产端仍然按普通 io.Writer 写 ZIP，上传端把 PipeReader 当作请求体。Go 官方文档说明，io.Pipe 的读写是同步匹配的，数据从 Write 直接交给对应 Read，没有内部缓冲。也就是说它不会偷偷囤积整包数据；当网络慢于生成速度时，写端会阻塞，形成自然背压。

先看各组件之间的静态关系

这个方案只有三个边界：生成边界负责产生记录并写 ZIP，管道边界负责连接 Reader 与 Writer，上传边界把 Reader 放进 HTTP Request Body。zip.Writer 不知道数据最终去了网络，http.Client 也不需要知道数据是现场生成的。

完整写法：边生成 ZIP 边作为请求体上传

下面的示例把一个记录生成函数写入 ZIP 中的 records.ndjson，随后直接上传。生成函数通过回调逐条提供记录，因此不要求调用方先创建完整切片。

package streamupload

import (
    "archive/zip"
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "net/http"
)

type Row struct {
    ID   int64  `json:"id"`
    Name string `json:"name"`
}

// RowGenerator 逐条产出记录，避免先构造完整数据集。
type RowGenerator func(yield func(Row) error) error

func writeZIP(dst io.Writer, generate RowGenerator) error {
    zw := zip.NewWriter(dst)
    entry, err := zw.Create("records.ndjson")
    if err != nil {
        return fmt.Errorf("create zip entry: %w", err)
    }

    enc := json.NewEncoder(entry)
    if err := generate(func(row Row) error {
        // 每次 Encode 都直接写向管道，不保存整包字节。
        return enc.Encode(row)
    }); err != nil {
        return fmt.Errorf("generate rows: %w", err)
    }

    // Close 会写出 ZIP 中央目录，必须在关闭 PipeWriter 前完成。
    if err := zw.Close(); err != nil {
        return fmt.Errorf("close zip: %w", err)
    }
    return nil
}

func UploadZIP(
    ctx context.Context,
    client *http.Client,
    endpoint string,
    generate RowGenerator,
) error {
    pr, pw := io.Pipe()

    req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, pr)
    if err != nil {
        // 请求未创建时关闭两端，避免遗留资源。
        _ = pr.Close()
        _ = pw.Close()
        return fmt.Errorf("create request: %w", err)
    }
    req.Header.Set("Content-Type", "application/zip")

    genErrCh := make(chan error, 1)
    go func() {
        genErr := writeZIP(pw, generate)
        // 生成失败会传到 HTTP 读取端；成功时等价于普通 Close。
        _ = pw.CloseWithError(genErr)
        genErrCh = 300 {
        return fmt.Errorf("upload status: %s", resp.Status)
    }
    return nil
}

这里最关键的不是 goroutine 本身，而是关闭责任。ZIP 生成完成后先关闭 zip.Writer，确保中央目录写完；再用 PipeWriter.CloseWithError 通知读取端成功或失败。若 client.Do 提前失败，关闭 PipeReader 会让仍在写入的生产者收到错误并退出。

卡住或得到截断包时按四层检查

我最初不适应的地方是：io.Pipe 没有内部缓冲，写端停住不一定是死锁，也可能只是上传端暂时没有继续读取。排查时把问题分层，比盯着一条 goroutine 堆栈更快。

错误与取消关系必须双向打通

CloseWithError 是这套方案能安全结束的关键。Go 文档说明，写端带错误关闭后，读端在数据读完后会收到该错误；读端带错误关闭后，后续写入会返回该错误。Context 又控制着外发请求从建连、发送到读取响应的整个生命周期。

如果只关闭写端而不等待 genErrCh，调用方可能漏掉生成失败；如果上传失败后不关闭读端，生产 goroutine 可能永远等不到消费者。生产和上传两条错误都要收集，必要时用 errors.Join 同时保留。

这三种情况不适合直接用 io.Pipe

1. 必须提前给出 Content-Length：流式生成通常不知道最终字节数。若服务端或签名协议要求固定长度，应使用原生分片上传，或先生成可重放的中间对象。
2. 请求必须自动重放：PipeReader 是一次性流，NewRequestWithContext 无法为它自动构造 GetBody。307/308 重定向或自动重试需要新的数据源，不能指望旧管道倒带。
3. 需要跨进程断点续传：io.Pipe 只连接当前进程里的生产者和消费者。超大任务若要从任意分片恢复，更适合对象存储的 multipart upload，并持久化分片编号与校验信息。

用四组场景做反向验证

这套代码不应只测“上传成功”。我会固定跑四组场景：

• 正常生成：服务端收到可解压 ZIP，记录数与生成数一致。
• 生成中途失败：生成函数返回自定义错误，UploadZIP 能收到该错误，HTTP 请求不会无限等待。
• 服务端提前拒绝：服务端立即返回错误或断开连接，生产 goroutine 能因为读端关闭而退出。
• Context 取消：上传过程中取消 Context，client.Do 返回，写端不残留阻塞 goroutine。

适合这套方案的场景，是“数据可顺序生成、目标端支持流式请求、失败后可以从头重跑”。它减少的是整包临时文件和峰值内存，不是让上传天然支持断点续传或自动重试。

常见问题

io.Pipe 会不会把数据全部缓存在内存？

不会。官方实现没有内部缓冲，写入会等待一个或多个读取完全消费当前数据。应用自身的 ZIP、JSON 或 HTTP 缓冲仍可能占少量内存，但不会由 io.Pipe 累积整包。

为什么 ZIP 必须先 Close 再关 PipeWriter？

zip.Writer.Close 会写入 ZIP 的中央目录。先关管道会让这部分数据无法发送，服务端拿到的包可能无法正常解压。

上传接口返回非 2xx 时是否能直接重试？

不能复用同一个 PipeReader。重试必须重新创建 pipe、请求和生成器；若生成不可重复，还需要可重放的存储或分片协议。

是否需要额外套 bufio.Writer？

通常先保持简单。额外缓冲可以减少小写入次数，但会改变内存与错误出现时机；只有测到大量微小 Write 成为瓶颈时再加入，并在取消路径中验证能否及时退出。

参考：Go 标准库 io、archive/zip 与 net/http 官方文档。本文配图均为原创静态结构图，不是运行截图或终端证据。