第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 大小。”

deffile_chunk_generator(filepath,chunk_size=8192):"""生成器：逐块读取文件"""withopen(filepath,"rb")asf:whilechunk:=f.read(chunk_size):yieldchunk# 流式上传——内存峰值 = 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可以包装生成器，实时显示进度条。”

fromtqdmimporttqdmdefprogress_generator(filepath,chunk_size=8192):total=os.path.getsize(filepath)withtqdm(total=total,unit="B",unit_scale=True)aspbar:withopen(filepath,"rb")asf:whilechunk:=f.read(chunk_size):yieldchunk pbar.update(len(chunk))

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

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

生活比喻	技术映射
水管送水（一点一点流）	流式上传（生成器 + chunked）
用卡车一次性运全部货物	Content-Length 一次性上传
分批快递（每次送一箱）	chunked 分块传输
快递单号实时跟踪	生成器中加进度回调
断掉的快递从中转站继续运	断点续传上传（服务端协调）

3. 项目实战

环境准备

pipinstallrequests requests-toolbelt tqdm

分步实现

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

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

importrequestsimportosimporttempfiledefread_file_chunks(filepath:str,chunk_size:int=65536):"""生成器：逐块读取文件（内存友好）"""withopen(filepath,"rb")asf:whileTrue:chunk=f.read(chunk_size)ifnotchunk:breakyieldchunkdefstream_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}")returnresp# ===== 测试（用 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)

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

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

importrequestsimportosimporttimefromtypingimportOptional,CallableclassProgressUploader:"""带进度追踪的流式上传器"""def__init__(self,chunk_size:int=65536):self.chunk_size=chunk_sizedef_progress_generator(self,filepath:str,callback:Optional[Callable]=None):"""生成器——逐块读取 + 进度回调"""total=os.path.getsize(filepath)uploaded=0start_time=time.perf_counter()withopen(filepath,"rb")asf:whilechunk:=f.read(self.chunk_size):yieldchunk uploaded+=len(chunk)ifcallback:elapsed=time.perf_counter()-start_time speed=uploaded/elapsedifelapsed>0else0pct=uploaded/total*100callback(uploaded,total,speed,elapsed,pct)defupload(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}ifextra_headers:headers.update(extra_headers)# 默认终端进度输出def_default_progress(uploaded,total,speed,elapsed,pct):mb_up=uploaded/1024**2mb_total=total/1024**2mb_s=speed/1024**2eta=(total-uploaded)/speedifspeed>0else0print(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_callbackor_default_progress resp=requests.put(url,data=self._progress_generator(filepath,cb),headers=headers,timeout=(10,600),)print()# 换行returnresp# ===== 使用演示 =====importtempfile# 创建一个 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 上传（文件 + 表单字段）。

fromrequests_toolbeltimportMultipartEncoder,MultipartEncoderMonitorimportrequestsimportosimporttempfiledefstream_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)withopen(filepath,"rb")asf:# 构建 multipart 字段form_fields={"file":(filename,f,"application/octet-stream"),}iffields:form_fields.update(fields)encoder=MultipartEncoder(fields=form_fields)# 可选：包装进度监控ifprogress_callback:def_progress(monitor):pct=monitor.bytes_read/monitor.len*100progress_callback(monitor.bytes_read,monitor.len,pct)monitor=MultipartEncoderMonitor(encoder,_progress)data=monitorelse:data=encoder resp=requests.post(url,data=data,headers={"Content-Type":encoder.content_type},timeout=(10,300),)returnresp# ===== 演示 =====tmp=tempfile.NamedTemporaryFile(delete=False,suffix=".dat")tmp.write(b"Hello Multipart Stream!"*1000)tmp.close()defshow_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"indata:print(f" 服务端收到文件:{data['files'].get('file','')[:50]}...")if"form"indata:print(f" 服务端收到字段:{data['form']}")os.unlink(tmp.name)

可能遇到的坑及解决方法

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

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

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

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

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

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

测试验证

importpytestimportrequestsimporttempfileimportosclassTestStreamUpload:"""验证流式上传功能"""deftest_generator_upload_to_httpbin(self):"""验证用生成器流式上传"""tmp=tempfile.NamedTemporaryFile(delete=False,suffix=".dat")tmp.write(b"test data "*100)tmp.close()defgen():withopen(tmp.name,"rb")asf:yieldf.read()resp=requests.put("https://httpbin.org/put",data=gen(),)assertresp.status_code==200os.unlink(tmp.name)deftest_chunked_large_data_no_oom(self):"""验证流式上传不会 OOM"""importtracemalloc tracemalloc.start()defgen():foriinrange(100):yieldb"A"*8192resp=requests.put("https://httpbin.org/put",data=gen(),)_,peak=tracemalloc.get_traced_memory()peak_mb=peak/1024**2# 内存峰值应该远小于 8KB*100 = 800KB（实际 ~10MB 包括库开销）assertresp.status_code==200print(f"\n 内存峰值:{peak_mb:.1f}MB")tracemalloc.stop()deftest_content_known_size_with_length(self):"""已知大小时传 Content-Length"""data=b"hello world"*100resp=requests.put("https://httpbin.org/put",data=data,headers={"Content-Length":str(len(data))},)assertresp.status_code==200

4. 项目总结

核心知识点

上传方式	内存占用	适用场景	Content-Type
`files={}`	高（文件大小）	小文件 + 表单	multipart/form-data
`data=bytes`	高（文件大小）	小文件纯二进制	自定义
`data=generator`	低（chunk_size）	大文件流式	chunked
MultipartEncoder	低（chunk_size）	大文件 + 表单	multipart/form-data

优点 & 缺点

对比维度	流式上传	传统上传
内存占用	极低（~KB 级）	等于文件大小
进度追踪	可在生成器中回调	困难（一次性）
断点续传	可实现	不可实现
代码复杂度	略高	低

适用场景

大文件上传（> 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。

思考题

设计题：设计一个"分片上传"方案——将 10GB 文件切成 10MB 的片段，每个片段独立上传。所有片段上传成功后，服务端合并。类似于 AWS S3 Multipart Upload。请设计协议：初始化上传（获取 upload_id）、上传分片（part_number + upload_id）、完成上传（列出所有分片的 ETag）、中止上传。客户端如何保证分片的完整性（MD5/ETag）？
可靠性题：流式上传期间网络断开，已经传输了 500MB/1GB。如何实现上传的断点续传？与下载不同，HTTP 没有标准的"Upload-Range"请求头。你需要设计一个客户端-服务端协议来支持这个功能。