第30章：上传视频不再“撑爆”服务器：手把手给 requests 接上“进度条水管”

1. 项目背景

业务场景

某视频 SaaS 平台允许用户通过 REST API 上传视频素材（最大 10GB）。初期实现是用 files 参数直接上传——requests.post(url, files={"video": open("movie.mp4", "rb")})。对于 50MB 以内的小视频正常运行，但用户开始上传 5GB+ 的电影素材时，服务器内存直接爆了——files 参数会把整个文件读入内存构造 multipart 请求体。

团队尝试用 data=open("movie.mp4", "rb") 加 Content-Type: video/mp4 直接传原始字节，绕过了 multipart 的内存问题。但新的问题出现了：服务器不知道文件什么时候传完（没有 Content-Length 头，因为文件大小未知或太大无法预先计算），导致连接在传完之前被超时断开。而且上传没有进度反馈——用户不知道已经传了多少了。

痛点

问题一：multipart 上传 vs 流式上传——内存占用的巨大差异。 files 参数会把整个 multipart 请求体（所有文件 + 所有 form 字段）在内存中组装完毕后再发送。一个 5GB 的视频意味着至少 5GB 的内存峰值。而流式上传使用生成器作为 data 参数，每次只读取一个文件块到内存、发送、释放——5GB 文件内存峰值可以控制在 8KB（chunk_size）。

问题二：Transfer-Encoding: chunked——不知道文件大小时的最佳选择。 HTTP 协议规定：如果请求有 body，必须通过 Content-Length 头告知服务端 body 的大小，或者使用 Transfer-Encoding: chunked 分块传输。当文件大小未知或太大无法预先计算时，chunked 编码是唯一选择。requests 在使用生成器作为 data 参数时会自动启用 chunked 编码。

问题三：大文件上传缺乏可观测性。 没有进度条、没有传输速度、没有预计剩余时间——运维无法判断上传是在正常进行还是卡死了。“已经传了 2 小时了，是服务器处理慢还是断网了？”——这个问题每周都会在运维群出现。

流式上传 vs 传统上传对比图：

传统上传（files 参数）:
内存: [████████████████] 5GB 全部加载
时间: 0s ───────────── 上传开始 ─────────────> 300s
→ 5GB 占用直到上传完成

流式上传（生成器 + chunked）:
内存: [█] 8KB chunk
时间: 0s ──> chunk1 ──> chunk2 ──> … ──> 300s
↑ 每块即时释放

2. 项目设计

小胖（看着服务器监控，CPU 和内存全线飙红）： “大师！视频上传服务又挂了——用户上传一个 8GB 的电影，服务器内存 16GB 直接飙到 95%，然后 OOM Killer 就来了。怎么才能上传大文件不爆内存？”

大师： “你的上传代码是怎么写的？requests.post(url, files={"video": f}) ？这种写法会把整个文件加载到内存来构造 multipart 请求体。8GB 文件 = 8GB+ 内存峰值。你需要流式上传。”

小胖（挠头）： “流式上传？我学过流式下载——用 stream=True + iter_content()。但上传怎么流式？”

大师： “用生成器作为 data 参数。当一个生成器被用作 data 时，requests 会自动切换到 Transfer-Encoding: chunked 模式——它不会一次性读取整个生成器，而是 yield 一块、发一块、yield 下一块、发下一块。内存占用就降到了 chunk 大小。”

def file_chunk_generator(filepath, chunk_size=8192):
"""生成器：逐块读取文件"""
with open(filepath, "rb") as f:
while chunk := f.read(chunk_size):
yield chunk

# 流式上传——内存峰值 = chunk_size (8KB)
requests.post(
"https://upload.example.com/videos",
data=file_chunk_generator("movie.mp4", 8192),
headers={"Content-Type": "video/mp4"},
)

小白（追问）： “那 Transfer-Encoding: chunked 和 Content-Length 有什么区别？什么情况下用哪个？”

大师： "这是 HTTP 协议中两个互斥的 body 长度指定方式：

• Content-Length: 524288000 — 提前告诉服务器：我要发 500MB，你准备好。服务器可以预先分配资源。
• Transfer-Encoding: chunked — 我不知道一共多大，一块一块发，最后发一个空 chunk 表示结束。

chunked 编码适合：文件大小未知（实时生成的数据）、文件太大无法一次性计算大小、流式数据（日志流、直播流）。但代价是服务器不能提前知道总大小（比如无法显示进度条给终端用户）。"

小胖： “那如果我想同时有进度条 + 流式上传呢？用户得知道上传了多少。”

大师： “你可以在生成器中加入回调——每读取一个 chunk 就调用回调函数报告进度。tqdm 可以包装生成器，实时显示进度条。”

from tqdm import tqdm

def progress_generator(filepath, chunk_size=8192):
total = os.path.getsize(filepath)
with tqdm(total=total, unit="B", unit_scale=True) as pbar:
with open(filepath, "rb") as f:
while chunk := f.read(chunk_size):
yield chunk
pbar.update(len(chunk))

小白（若有所思）： “那如果上传中断了呢？用户得从 50% 重新开始吗？”

大师： “这就是断点续传上传——你需要服务端支持。流程是：客户端先询问服务端’你已经收到了多少’（GET /upload/status?file_id=xxx），服务端返回已接收的字节数。客户端从那个位置开始继续上传——类似于下载的 Range 头，但上传没有标准的 Range 头，通常用自定义头或 Upload-Offset。”

3. 项目实战

环境准备

pip install requests requests-toolbelt tqdm

分步实现

步骤一：生成器流式上传——基础版

目标：实现最简单的流式上传，理解 chunked 编码。

import requests
import os
import tempfile

def read_file_chunks(filepath: str, chunk_size: int = 65536):
"""生成器：逐块读取文件（内存友好）"""
with open(filepath, "rb") as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
yield chunk

def stream_upload_raw(url: str, filepath: str,
content_type: str = "application/octet-stream"):
"""流式上传——用生成器作为 data 参数

注意：这种方式发送的是原始二进制 body，不走 multipart 封装。
适合服务端接受原始二进制上传的场景（如 S3 presigned URL）。
"""
file_size = os.path.getsize(filepath)
print(f"文件: {os.path.basename(filepath)} ({file_size:,} bytes)")

# 使用生成器——requests 自动启用 chunked 编码
resp = requests.put(
url,
data=read_file_chunks(filepath),
headers={
"Content-Type": content_type,
"Content-Length": str(file_size), # 如果知道大小就传
},
timeout=(10, 300),
)

print(f"上传完成: {resp.status_code}")
return resp

# ===== 测试（用 httpbin） =====
# httpbin.org/put 会回显请求体大小
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".bin")
tmp.write(b"A" * 100000) # 100KB 测试文件
tmp.close()

resp = stream_upload_raw("https://httpbin.org/put", tmp.name)
data = resp.json()
print(f"服务端收到的数据大小: {len(data.get('data', ''))} 字节")

os.unlink(tmp.name)

步骤二：带进度条的流式上传

目标：实现带实时进度、传输速度的大文件上传。

import requests
import os
import time
from typing import Optional, Callable

class ProgressUploader:
"""带进度追踪的流式上传器"""

def __init__(self, chunk_size: int = 65536):
self.chunk_size = chunk_size

def _progress_generator(self, filepath: str,
callback: Optional[Callable] = None):
"""生成器——逐块读取 + 进度回调"""
total = os.path.getsize(filepath)
uploaded = 0
start_time = time.perf_counter()

with open(filepath, "rb") as f:
while chunk := f.read(self.chunk_size):
yield chunk
uploaded += len(chunk)

if callback:
elapsed = time.perf_counter() – start_time
speed = uploaded / elapsed if elapsed > 0 else 0
pct = uploaded / total * 100
callback(uploaded, total, speed, elapsed, pct)

def upload(
self,
url: str,
filepath: str,
content_type: str = "application/octet-stream",
extra_headers: dict = None,
progress_callback: Optional[Callable] = None,
) –> requests.Response:
"""上传大文件——带进度回调"""

headers = {"Content-Type": content_type}
if extra_headers:
headers.update(extra_headers)

# 默认终端进度输出
def _default_progress(uploaded, total, speed, elapsed, pct):
mb_up = uploaded / 1024**2
mb_total = total / 1024**2
mb_s = speed / 1024**2
eta = (total – uploaded) / speed if speed > 0 else 0
print(f"\\r 进度: {mb_up:.1f}/{mb_total:.1f}MB "
f"({pct:.0f}%) {mb_s:.1f}MB/s "
f"ETA: {eta:.0f}s ", end="", flush=True)

cb = progress_callback or _default_progress

resp = requests.put(
url,
data=self._progress_generator(filepath, cb),
headers=headers,
timeout=(10, 600),
)
print() # 换行
return resp

# ===== 使用演示 =====
import tempfile

# 创建一个 1MB 的测试文件
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".bin")
tmp.write(b"X" * (1024 * 1024))
tmp.close()

uploader = ProgressUploader(chunk_size=16384)
resp = uploader.upload("https://httpbin.org/put", tmp.name)
print(f"上传结果: {resp.status_code}")

os.unlink(tmp.name)

步骤三：requests-toolbelt MultipartEncoder——流式 multipart

目标：使用 requests-toolbelt 实现流式 multipart 上传（文件 + 表单字段）。

from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
import requests
import os
import tempfile

def stream_multipart_upload(
url: str,
filepath: str,
fields: dict = None,
progress_callback=None,
):
"""流式 multipart 上传——支持文件 + 表单字段

与 requests 的 files 参数不同：
– files 参数：整个请求体在内存中组装
– MultipartEncoder：流式读取文件，逐块组装
"""

filename = os.path.basename(filepath)
file_size = os.path.getsize(filepath)

with open(filepath, "rb") as f:
# 构建 multipart 字段
form_fields = {
"file": (filename, f, "application/octet-stream"),
}
if fields:
form_fields.update(fields)

encoder = MultipartEncoder(fields=form_fields)

# 可选：包装进度监控
if progress_callback:
def _progress(monitor):
pct = monitor.bytes_read / monitor.len * 100
progress_callback(monitor.bytes_read, monitor.len, pct)

monitor = MultipartEncoderMonitor(encoder, _progress)
data = monitor
else:
data = encoder

resp = requests.post(
url,
data=data,
headers={"Content-Type": encoder.content_type},
timeout=(10, 300),
)

return resp

# ===== 演示 =====
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".dat")
tmp.write(b"Hello Multipart Stream!" * 1000)
tmp.close()

def show_progress(uploaded, total, pct):
print(f"\\r multipart 上传: {uploaded:,}/{total:,} ({pct:.0f}%)", end="")

resp = stream_multipart_upload(
"https://httpbin.org/post",
tmp.name,
fields={"author": "小胖", "department": "IT"},
progress_callback=show_progress,
)
print(f"\\n结果: {resp.status_code}")

data = resp.json()
if "files" in data:
print(f" 服务端收到文件: {data['files'].get('file', '')[:50]}…")
if "form" in data:
print(f" 服务端收到字段: {data['form']}")

os.unlink(tmp.name)

可能遇到的坑及解决方法

坑1：生成器只能迭代一次

# 如果生成器在第一次迭代后被重用，不会产生任何数据
# 因为生成器已经"耗尽"了
# 解决：每次上传前创建新的生成器实例
generator = read_file_chunks("file.bin")
requests.put(url, data=generator) # 第一次 OK
requests.put(url, data=generator) # ❌ 第二次不发送任何数据！

坑2：chunked 编码与 Content-Length 的冲突

# 如果同时设置了 Content-Length 和使用生成器作为 data
# urllib3 可能忽略生成器的 chunked 行为，采用 Content-Length
# 如果你的 Content-Length 不准确，服务端会等到超时

坑3：流式上传不支持自动重试

# 与普通请求不同，生成器不能被重试（因为无法"重置"）
# 如果上传失败，需要从文件开头重新创建生成器
# 这也就是为什么流式上传的 retry 需要客户端配合

测试验证

import pytest
import requests
import tempfile
import os

class TestStreamUpload:
"""验证流式上传功能"""

def test_generator_upload_to_httpbin(self):
"""验证用生成器流式上传"""
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".dat")
tmp.write(b"test data " * 100)
tmp.close()

def gen():
with open(tmp.name, "rb") as f:
yield f.read()

resp = requests.put(
"https://httpbin.org/put",
data=gen(),
)
assert resp.status_code == 200
os.unlink(tmp.name)

def test_chunked_large_data_no_oom(self):
"""验证流式上传不会 OOM"""
import tracemalloc
tracemalloc.start()

def gen():
for i in range(100):
yield b"A" * 8192

resp = requests.put(
"https://httpbin.org/put",
data=gen(),
)
_, peak = tracemalloc.get_traced_memory()
peak_mb = peak / 1024**2
# 内存峰值应该远小于 8KB*100 = 800KB（实际 ~10MB 包括库开销）
assert resp.status_code == 200
print(f"\\n 内存峰值: {peak_mb:.1f}MB")
tracemalloc.stop()

def test_content_known_size_with_length(self):
"""已知大小时传 Content-Length"""
data = b"hello world" * 100
resp = requests.put(
"https://httpbin.org/put",
data=data,
headers={"Content-Length": str(len(data))},
)
assert resp.status_code == 200

4. 项目总结

核心知识点

优点 & 缺点

适用场景

• 大文件上传（> 100MB）：视频、数据集、备份文件
• 实时数据流：日志流、IoT 传感器数据
• 内存受限环境：容器、边缘设备
• 未知大小的数据：实时生成的数据流

注意事项

• 生成器只能迭代一次——每次上传需创建新的生成器
• Content-Length 和 chunked 编码互斥
• requests-toolbelt 的 MultipartEncoder 支持文件 + 表单字段的流式上传
• 流式上传失败无法自动重试（生成器无法重置）

常见踩坑经验

案例一：5GB 视频上传导致服务器 OOM。 视频上传 API 收到 5GB 视频文件，服务端用 Flask 的 request.files['video'].read() 把整个文件读到内存做病毒扫描。内存不足触发 OOM。根因：服务端也未实现流式处理。修复：服务端用流式解析（request.stream），客户端用 MultipartEncoder 分块上传。

案例二：生成器重用的静默失败。 调试代码时先生成了一次生成器并打印了第一块数据，然后用同一个生成器传给了 requests。结果上传成功但文件是空的——因为生成器在 print 时已经消耗掉了第一块，而后续读取时文件指针已经被移动。根因：生成器被部分消耗后传给 requests。修复：每次上传重新创建生成器。

案例三：Content-Length 不准导致上传挂起。 文件修改脚本在计算完文件大小后、开始上传前，另一个进程修改了文件（追加了数据）。Content-Length 声明的是旧大小，但实际传输的数据更多。服务器严格按照 Content-Length 读取，超过的部分被当成了下一个请求的请求头——导致后续请求解析失败。根因：文件在上传期间被修改。修复：上传前锁定文件或使用 chunked 编码避免依赖准确的 Content-Length。

思考题

延伸阅读与资源

Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地 后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战 10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用 后端工程师转型AI第一课-Ollama 与私有化大模型实战

大型语言模型(LLM) vLLM 高性能推理落地实战

Agent开发之LlamaIndex 实战修炼与源码进阶

大语言模型Transformers 实战修炼与源码剖析