Go io.Pipe 如何把编码器输出接到上传请求

我在做大批量数据上传时，最不愿意看到的写法是先 json.Marshal 成一个很大的 []byte，再把它交给 HTTP 客户端。数据量一上来，内存峰值和等待时间都会一起变得难看。更合适的连接方式是：io.Pipe 的写端交给编码器，读端直接作为请求体；编码器在一个 goroutine 里持续写，请求在另一个方向持续读。

关键不是把编码器“塞进”请求，而是把两者接成一条有背压的 Reader/Writer 通道。正常完成时关闭 PipeWriter 让请求读到 EOF；编码失败时用 CloseWithError 把错误传给读取端，并让请求使用可取消的 context。

先把 io.Pipe 两端接对

io.Pipe() 返回 *io.PipeReader 和 *io.PipeWriter。它们不是一个带容量的队列，而是同步交接：写端写入的数据要被读端消费，写调用才会继续。因此它适合把“边编码边上传”连起来，也意味着远端变慢时编码器会自然受到背压。

职责可以先按这张表固定下来：

下面的连接关系是文章中的操作示意，不代表已经在本机执行。

请求先启动，编码器再开始写

最容易踩的坑是按“先编码、后发请求”思考。因为管道没有内部缓冲，编码 goroutine 必须和 client.Do 并行。示例把业务对象逐条编码成 JSON Lines；如果接口要求一个完整 JSON 数组，可把数组首尾符号也写入同一个 writer，原则不变。

package main

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

type Item struct {
    ID    int    `json:"id"`
    Label string `json:"label"`
}

func upload(ctx context.Context, client *http.Client, uploadURL string, items []Item) error {
    pr, pw := io.Pipe()

    // 编码器在独立 goroutine 中写入，避免第一笔写入等不到请求读取。
    encodeErr := make(chan error, 1)
    go func() {
        enc := json.NewEncoder(pw)
        for _, item := range items {
            // Encode 会写入一条 JSON 并追加换行，适合逐条发送的接口。
            if err := enc.Encode(item); err != nil {
                _ = pw.CloseWithError(fmt.Errorf("encode item %d: %w", item.ID, err))
                encodeErr = 300 {
        _, _ = io.Copy(io.Discard, resp.Body)
        return fmt.Errorf("upload returned %s", resp.Status)
    }
    if err := 

这里没有为示例硬编码上传地址，调用方应传入自己的接口 URL。NewRequestWithContext 会把请求体作为 io.Reader 使用；同时 context 可以覆盖建连、发送请求以及读取响应的整个生命周期。

正常 EOF 和编码失败必须分开传递

正常路径只需要 pw.Close()，读端最终得到 EOF。编码器出错时若仍然普通关闭，上传端可能把“半截数据”当成正常结束，所以应使用 pw.CloseWithError(err)。反过来，HTTP 客户端已经失败时，调用 pr.CloseWithError(err) 可以让生产端正在等待的 Write 返回。

判断结果时建议保留三类信号：

• 返回了响应但状态码不是 2xx：这是服务端业务或协议拒绝，不等同于 Go 编码失败。
• client.Do 返回错误：通常要看 context、连接、TLS 或传输层，并通知 PipeWriter 停止。
• 响应成功但编码 goroutine 报错：请求可能已经发送了一部分，是否重试要由接口幂等性和服务端协议决定。

收尾时检查响应、取消和重试边界

生产环境里我会把收尾顺序写进检查清单：先确保 resp.Body.Close() 一定执行；再记录 HTTP 状态与响应摘要；最后决定是否重试。上传请求不是天然可重试的，若编码输出已经产生副作用或服务端不支持幂等键，盲目重试可能造成重复数据。

如果数据本身已经完整落盘、接口又要求可重放，临时文件加 os.File 往往比 Pipe 更容易重试和定位。Pipe 的优势是流式和低额外内存，不是替代所有上传队列。

常见问题

io.Pipe 会缓存多少数据？

它没有内部缓冲；写入与读取同步匹配。需要可控缓冲时，应显式增加缓冲层，并重新评估内存上限和背压。

为什么编码器要放 goroutine？

因为写端可能在第一笔写入时等待读端。请求先进入 Client.Do，读端才会持续消费 Pipe。

请求返回 2xx 就能忽略编码错误吗？

不能。服务端可能已接收部分内容；仍要等待并检查编码 goroutine 的结果，再依据接口幂等规则决定后续动作。