本文介绍如何通过包装 http.ResponseWriter 实现对 HTTP 流式响应过程中已发送字节数与瞬时/平均下载速度的精确监控，适用于大文件分块传输、带宽自适应调度等场景。
本文介绍如何通过包装 `http.responsewriter` 实现对 http 流式响应过程中已发送字节数与瞬时/平均下载速度的精确监控，适用于大文件分块传输、带宽自适应调度等场景。
在构建流式下载服务（如对象存储的 chunked blob 下载）时，单纯调用 w.Write() 无法获知数据何时真正写入网络连接、是否被客户端接收或当前吞吐瓶颈在哪。Go 标准库的 http.ResponseWriter 是一个接口，不暴露底层缓冲状态或已写入字节数——这意味着你无法直接查询“还有多少数据排队等待发送”或“过去 1 秒发出了多少字节”。
解决思路是：实现一个中间层 ResponseWriter 包装器（wrapper），在每次 Write() 调用时记录字节数，并结合时间戳计算速率。以下是一个轻量、生产就绪的示例实现：
import (
"io"
"net/http"
"time"
)
type MeasuredWriter struct {
http.ResponseWriter
written int64
start time.Time
last time.Time
}
func NewMeasuredWriter(w http.ResponseWriter) *MeasuredWriter {
return &MeasuredWriter{
ResponseWriter: w,
start: time.Now(),
last: time.Now(),
}
}
func (mw *MeasuredWriter) Write(p []byte) (int, error) {
n, err := mw.ResponseWriter.Write(p)
mw.written += int64(n)
mw.last = time.Now()
return n, err
}
// BytesWritten 返回累计写入字节数
func (mw *MeasuredWriter) BytesWritten() int64 {
return mw.written
}
// AvgSpeed returns average speed in bytes/sec since start
func (mw *MeasuredWriter) AvgSpeed() float64 {
elapsed := time.Since(mw.start).Seconds()
if elapsed == 0 {
return 0
}
return float64(mw.written) / elapsed
}
// InstantSpeed returns approximate instantaneous speed (last 500ms window)
func (mw *MeasuredWriter) InstantSpeed() float64 {
elapsed := time.Since(mw.last).Seconds()
// For simplicity: use avg over last write interval — in practice,
// maintain a sliding window of recent writes for true instant rate
if elapsed < 0.5 {
return mw.AvgSpeed()
}
return 0
}在你的 handler 中使用它：
func downloadHandler(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
mw := NewMeasuredWriter(w)
// 设置必要 header（务必在第一次 Write 前）
mw.Header().Set("Content-Disposition", `attachment; filename="blob.dat"`)
mw.Header().Set("Content-Type", "application/octet-stream")
f, err := os.Open("somefile.txt")
if err != nil {
http.Error(mw, "file not found", http.StatusNotFound)
return
}
defer f.Close()
buf := make([]byte, 32*1024) // 推荐 32KB 缓冲区以平衡延迟与内存
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
go func() {
for range ticker.C {
log.Printf("Download progress: %d bytes | Avg speed: %.2f KB/s",
mw.BytesWritten(), mw.AvgSpeed()/1024)
}
}()
for {
n, err := f.Read(buf)
if n > 0 {
if _, writeErr := mw.Write(buf[:n]); writeErr != nil {
log.Printf("Write error: %v", writeErr)
return // 客户端断连等异常
}
}
if err == io.EOF {
break
}
if err != nil {
http.Error(mw, "read error", http.StatusInternalServerError)
return
}
}
}✅ 关键注意事项：
- 必须在首次
Write()前设置Header()，否则WriteHeader()会被隐式触发，导致无法再修改 header； -
http.ResponseWriter的实际写入受 HTTP/1.1 缓冲、TCP 窗口、客户端接收能力等多层影响，Write()返回成功仅表示数据进入 Go 的bufio.Writer，不代表已送达客户端； - 如需更精准的“瞬时速率”，建议维护一个最近 5–10 次
Write()的时间/字节数环形缓冲区，而非依赖单次间隔； - 对于高并发场景，避免在
Write()中执行耗时操作（如日志打印），可改用 goroutine + channel 异步上报； - 第三方库如
github.com/miolini/datacounter提供了开箱即用的计数器，但其本身不内置速率计算，仍需自行封装逻辑。
通过该模式，你不仅能实时感知客户端下载能力，还可据此动态调整后端 chunk 加载节奏（例如：检测到速率持续低于阈值时，暂缓下一块读取，避免内存积压），真正实现“按需拉取”的弹性流式传输。
