Go如何通过实现io.Reader接口取消客户端表单上传

原来靠time.Sleep设固定时长等上传到指定进度再取消的方案，问题在于休眠时长完全靠猜：设短了还没传到指定进度就触发取消，设长了要多等无意义的时间，大文件、弱网场景下偏差极大。
用自定义io.Reader的方案可以完全规避这个问题，因为HTTP客户端发送请求体时，会循环调用Reader的Read方法拉取要发送的字节，你可以直接在Read方法里精确判断当前已发送的字节数，到阈值立刻触发取消，不需要靠休眠猜时间。

• 给自定义Reader绑定context的cancel函数，实时记录已发送的字节偏移量
• 每次Read被调用、填充完要发送的字节后，判断偏移量是否达到终止阈值，达到就立刻调用cancel()
• 用io.Pipe把multipart表单的写入流和HTTP请求的读取流对接，不需要提前把整个文件读进内存，大文件上传内存占用稳定
• 触发cancel后直接在Read方法返回错误，让HTTP层立刻终止请求发送，不会有多余的网络传输
• 如果需要由用户操作（比如点击取消按钮）触发终止，不需要等字节数到阈值，直接在按钮回调里调用同一个cancel函数即可，逻辑完全通用

原有实现存在几个影响可靠性的问题，已经一并修复：

• 原CancelReader用值接收者，内部字段修改不会作用到原对象，并发场景下会有数据竞争，改成指针接收者
• 原逻辑提前把整个本地文件读进bytes.Buffer，大文件会直接占满内存，改为流式读写
• 原逻辑触发cancel后没有及时返回错误，HTTP客户端还会继续尝试读取后续数据，取消有延迟
• 原代码里测试用的随机生成字节逻辑替换为真实文件内容读取

客户端代码 #

package main

import (
	"context"
	"io"
	"log"
	"mime/multipart"
	"net/http"
	"os"
)

// CancelReader 支持主动取消的请求体Reader
type CancelReader struct {
	underlying io.Reader // 底层实际的数据源（multipart表单流）
	cancel     func()    // 绑定context的取消函数
	offset     int64     // 已发送的字节数
	stopAt     int64     // 触发取消的字节阈值
	canceled   bool      // 取消标记，避免重复调用cancel
}

func (c *CancelReader) Read(p []byte) (n int, err error) {
	// 已经取消过直接返回错误
	if c.canceled {
		return 0, context.Canceled
	}
	// 从底层流读数据
	n, err = c.underlying.Read(p)
	c.offset += int64(n)
	log.Printf("已发送字节数: %d, 取消阈值: %d\n", c.offset, c.stopAt)
	// 达到取消阈值，主动触发取消
	if c.offset >= c.stopAt && !c.canceled {
		c.cancel()
		c.canceled = true
		return n, context.Canceled
	}
	return n, err
}

const (
	filename  = "/Users/jimyag/Downloads/aDrive.dmg" // 替换为本地文件实际路径
	targetUrl = "http://localhost:9999/upload"
)

func main() {
	// 初始化带取消的context
	cx, cancel := context.WithCancel(context.Background())

	// 用Pipe对接multipart写入和请求读取，不需要全量加载文件到内存
	pr, pw := io.Pipe()
	bodyWriter := multipart.NewWriter(pw)

	// 异步写multipart表单内容
	go func() {
		defer pw.Close()
		defer bodyWriter.Close()

		// 创建表单文件字段
		fileWriter, err := bodyWriter.CreateFormFile("uploadfile", filename)
		if err != nil {
			log.Printf("创建表单文件字段失败: %v", err)
			pw.CloseWithError(err)
			return
		}

		// 打开本地文件
		fh, err := os.Open(filename)
		if err != nil {
			log.Printf("打开本地文件失败: %v", err)
			pw.CloseWithError(err)
			return
		}
		defer fh.Close()

		// 拷贝文件内容到multipart流
		_, err = io.Copy(fileWriter, fh)
		if err != nil {
			log.Printf("写入文件内容失败: %v", err)
			pw.CloseWithError(err)
			return
		}
	}()

	// 包装成支持主动取消的Reader，示例配置为发送50KB就自动取消
	cancelReader := &CancelReader{
		underlying: pr,
		cancel:     cancel,
		offset:     0,
		stopAt:     1024 * 50,
		canceled:   false,
	}

	// 构建请求
	req, err := http.NewRequest(http.MethodPost, targetUrl, cancelReader)
	if err != nil {
		log.Printf("构建请求失败: %v", err)
		return
	}
	req.Header.Set("Content-Type", bodyWriter.FormDataContentType())
	req = req.WithContext(cx)

	// 发送请求
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		log.Printf("请求终止: %v", err)
		return
	}
	defer resp.Body.Close()
	log.Println("请求正常完成")
}

服务端代码 #

package main

import (
	"errors"
	"io"
	"log"
	"net/http"
)

func main() {
	http.HandleFunc("/upload", func(w http.ResponseWriter, r *http.Request) {
		// 配置32MB内存上限，避免大文件占满服务端内存
		err := r.ParseMultipartForm(32 << 20)
		if err != nil {
			if r.Context().Err() != nil {
				log.Printf("客户端主动终止请求，context错误: %v", r.Context().Err())
			} else {
				log.Printf("解析表单失败: %v", err)
			}
			return
		}

		file, _, err := r.FormFile("uploadfile")
		if err != nil {
			log.Printf("获取上传文件失败: %v", err)
			return
		}
		defer file.Close()

		buf := make([]byte, 1024)
		totalSize := 0
		for {
			n, err := file.Read(buf)
			totalSize += n
			if err != nil {
				if errors.Is(err, io.EOF) {
					log.Printf("文件上传完成，总大小: %d字节", totalSize)
				} else {
					log.Printf("读取上传内容中断，已接收大小: %d字节，错误: %v", totalSize, err)
				}
				break
			}
		}
	})

	log.Println("服务启动，监听9999端口")
	err := http.ListenAndServe(":9999", nil)
	if err != nil {
		log.Printf("服务启动失败: %v", err)
	}
}

• 全程没有任何time.Sleep硬编码等待逻辑，取消时机完全由实际发送的字节数决定，不受网络波动、文件大小影响
• 内存占用稳定，不管上传多大的文件，都不会把整个文件加载到内存
• 触发取消后客户端立刻终止请求发送，服务端会立刻收到连接中断的信号，没有多余的流量浪费

内容的提问来源于stack exchange，提问作者Jimyag