1. 项目背景业务场景某视频 SaaS 平台允许用户通过 REST API 上传视频素材最大 10GB。初期实现是用files参数直接上传——requests.post(url, files{video: open(movie.mp4, rb)})。对于 50MB 以内的小视频正常运行但用户开始上传 5GB 的电影素材时服务器内存直接爆了——files参数会把整个文件读入内存构造 multipart 请求体。团队尝试用dataopen(movie.mp4, rb)加Content-Type: video/mp4直接传原始字节绕过了 multipart 的内存问题。但新的问题出现了服务器不知道文件什么时候传完没有Content-Length头因为文件大小未知或太大无法预先计算导致连接在传完之前被超时断开。而且上传没有进度反馈——用户不知道已经传了多少了。痛点问题一multipart 上传 vs 流式上传——内存占用的巨大差异。files参数会把整个 multipart 请求体所有文件 所有 form 字段在内存中组装完毕后再发送。一个 5GB 的视频意味着至少 5GB 的内存峰值。而流式上传使用生成器作为data参数每次只读取一个文件块到内存、发送、释放——5GB 文件内存峰值可以控制在 8KBchunk_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 内存峰值。你需要流式上传。”小胖挠头“流式上传我学过流式下载——用streamTrueiter_content()。但上传怎么流式”大师“用生成器作为data参数。当一个生成器被用作data时requests 会自动切换到Transfer-Encoding: chunked模式——它不会一次性读取整个生成器而是 yield 一块、发一块、yield 下一块、发下一块。内存占用就降到了 chunk 大小。”deffile_chunk_generator(filepath,chunk_size8192):生成器逐块读取文件withopen(filepath,rb)asf:whilechunk:f.read(chunk_size):yieldchunk# 流式上传——内存峰值 chunk_size (8KB)requests.post(https://upload.example.com/videos,datafile_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_size8192):totalos.path.getsize(filepath)withtqdm(totaltotal,unitB,unit_scaleTrue)aspbar:withopen(filepath,rb)asf:whilechunk:f.read(chunk_size):yieldchunk pbar.update(len(chunk))小白若有所思“那如果上传中断了呢用户得从 50% 重新开始吗”大师“这就是断点续传上传——你需要服务端支持。流程是客户端先询问服务端’你已经收到了多少’GET /upload/status?file_idxxx服务端返回已接收的字节数。客户端从那个位置开始继续上传——类似于下载的Range头但上传没有标准的 Range 头通常用自定义头或 Upload-Offset。”生活比喻技术映射水管送水一点一点流流式上传生成器 chunked用卡车一次性运全部货物Content-Length 一次性上传分批快递每次送一箱chunked 分块传输快递单号实时跟踪生成器中加进度回调断掉的快递从中转站继续运断点续传上传服务端协调3. 项目实战环境准备pipinstallrequests requests-toolbelt tqdm分步实现步骤一生成器流式上传——基础版目标实现最简单的流式上传理解 chunked 编码。importrequestsimportosimporttempfiledefread_file_chunks(filepath:str,chunk_size:int65536):生成器逐块读取文件内存友好withopen(filepath,rb)asf:whileTrue:chunkf.read(chunk_size)ifnotchunk:breakyieldchunkdefstream_upload_raw(url:str,filepath:str,content_type:strapplication/octet-stream):流式上传——用生成器作为 data 参数 注意这种方式发送的是原始二进制 body不走 multipart 封装。 适合服务端接受原始二进制上传的场景如 S3 presigned URL。 file_sizeos.path.getsize(filepath)print(f文件:{os.path.basename(filepath)}({file_size:,}bytes))# 使用生成器——requests 自动启用 chunked 编码resprequests.put(url,dataread_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 会回显请求体大小tmptempfile.NamedTemporaryFile(deleteFalse,suffix.bin)tmp.write(bA*100000)# 100KB 测试文件tmp.close()respstream_upload_raw(https://httpbin.org/put,tmp.name)dataresp.json()print(f服务端收到的数据大小:{len(data.get(data,))}字节)os.unlink(tmp.name)步骤二带进度条的流式上传目标实现带实时进度、传输速度的大文件上传。importrequestsimportosimporttimefromtypingimportOptional,CallableclassProgressUploader:带进度追踪的流式上传器def__init__(self,chunk_size:int65536):self.chunk_sizechunk_sizedef_progress_generator(self,filepath:str,callback:Optional[Callable]None):生成器——逐块读取 进度回调totalos.path.getsize(filepath)uploaded0start_timetime.perf_counter()withopen(filepath,rb)asf:whilechunk:f.read(self.chunk_size):yieldchunk uploadedlen(chunk)ifcallback:elapsedtime.perf_counter()-start_time speeduploaded/elapsedifelapsed0else0pctuploaded/total*100callback(uploaded,total,speed,elapsed,pct)defupload(self,url:str,filepath:str,content_type:strapplication/octet-stream,extra_headers:dictNone,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_upuploaded/1024**2mb_totaltotal/1024**2mb_sspeed/1024**2eta(total-uploaded)/speedifspeed0else0print(f\r 进度:{mb_up:.1f}/{mb_total:.1f}MB f({pct:.0f}%){mb_s:.1f}MB/s fETA:{eta:.0f}s ,end,flushTrue)cbprogress_callbackor_default_progress resprequests.put(url,dataself._progress_generator(filepath,cb),headersheaders,timeout(10,600),)print()# 换行returnresp# 使用演示 importtempfile# 创建一个 1MB 的测试文件tmptempfile.NamedTemporaryFile(deleteFalse,suffix.bin)tmp.write(bX*(1024*1024))tmp.close()uploaderProgressUploader(chunk_size16384)respuploader.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:dictNone,progress_callbackNone,):流式 multipart 上传——支持文件 表单字段 与 requests 的 files 参数不同 - files 参数整个请求体在内存中组装 - MultipartEncoder流式读取文件逐块组装 filenameos.path.basename(filepath)file_sizeos.path.getsize(filepath)withopen(filepath,rb)asf:# 构建 multipart 字段form_fields{file:(filename,f,application/octet-stream),}iffields:form_fields.update(fields)encoderMultipartEncoder(fieldsform_fields)# 可选包装进度监控ifprogress_callback:def_progress(monitor):pctmonitor.bytes_read/monitor.len*100progress_callback(monitor.bytes_read,monitor.len,pct)monitorMultipartEncoderMonitor(encoder,_progress)datamonitorelse:dataencoder resprequests.post(url,datadata,headers{Content-Type:encoder.content_type},timeout(10,300),)returnresp# 演示 tmptempfile.NamedTemporaryFile(deleteFalse,suffix.dat)tmp.write(bHello Multipart Stream!*1000)tmp.close()defshow_progress(uploaded,total,pct):print(f\r multipart 上传:{uploaded:,}/{total:,}({pct:.0f}%),end)respstream_multipart_upload(https://httpbin.org/post,tmp.name,fields{author:小胖,department:IT},progress_callbackshow_progress,)print(f\n结果:{resp.status_code})dataresp.json()iffilesindata:print(f 服务端收到文件:{data[files].get(file,)[:50]}...)ifformindata:print(f 服务端收到字段:{data[form]})os.unlink(tmp.name)可能遇到的坑及解决方法坑1生成器只能迭代一次# 如果生成器在第一次迭代后被重用不会产生任何数据# 因为生成器已经耗尽了# 解决每次上传前创建新的生成器实例generatorread_file_chunks(file.bin)requests.put(url,datagenerator)# 第一次 OKrequests.put(url,datagenerator)# ❌ 第二次不发送任何数据坑2chunked 编码与 Content-Length 的冲突# 如果同时设置了 Content-Length 和使用生成器作为 data# urllib3 可能忽略生成器的 chunked 行为采用 Content-Length# 如果你的 Content-Length 不准确服务端会等到超时坑3流式上传不支持自动重试# 与普通请求不同生成器不能被重试因为无法重置# 如果上传失败需要从文件开头重新创建生成器# 这也就是为什么流式上传的 retry 需要客户端配合测试验证importpytestimportrequestsimporttempfileimportosclassTestStreamUpload:验证流式上传功能deftest_generator_upload_to_httpbin(self):验证用生成器流式上传tmptempfile.NamedTemporaryFile(deleteFalse,suffix.dat)tmp.write(btest data *100)tmp.close()defgen():withopen(tmp.name,rb)asf:yieldf.read()resprequests.put(https://httpbin.org/put,datagen(),)assertresp.status_code200os.unlink(tmp.name)deftest_chunked_large_data_no_oom(self):验证流式上传不会 OOMimporttracemalloc tracemalloc.start()defgen():foriinrange(100):yieldbA*8192resprequests.put(https://httpbin.org/put,datagen(),)_,peaktracemalloc.get_traced_memory()peak_mbpeak/1024**2# 内存峰值应该远小于 8KB*100 800KB实际 ~10MB 包括库开销assertresp.status_code200print(f\n 内存峰值:{peak_mb:.1f}MB)tracemalloc.stop()deftest_content_known_size_with_length(self):已知大小时传 Content-Lengthdatabhello world*100resprequests.put(https://httpbin.org/put,datadata,headers{Content-Length:str(len(data))},)assertresp.status_code2004. 项目总结核心知识点上传方式内存占用适用场景Content-Typefiles{}高文件大小小文件 表单multipart/form-datadatabytes高文件大小小文件纯二进制自定义datagenerator低chunk_size大文件流式chunkedMultipartEncoder低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请求头。你需要设计一个客户端-服务端协议来支持这个功能。延伸阅读与资源Milvus向量数据库实战修炼从 0 到 1精通向量检索与生产落地后端工程师的 AI 转型第一课Ollama 与私有化大模型实战10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析
第30章：上传视频不再“撑爆”服务器：手把手给 requests 接上“进度条水管”
发布时间：2026/7/13 9:33:41
1. 项目背景业务场景某视频 SaaS 平台允许用户通过 REST API 上传视频素材最大 10GB。初期实现是用files参数直接上传——requests.post(url, files{video: open(movie.mp4, rb)})。对于 50MB 以内的小视频正常运行但用户开始上传 5GB 的电影素材时服务器内存直接爆了——files参数会把整个文件读入内存构造 multipart 请求体。团队尝试用dataopen(movie.mp4, rb)加Content-Type: video/mp4直接传原始字节绕过了 multipart 的内存问题。但新的问题出现了服务器不知道文件什么时候传完没有Content-Length头因为文件大小未知或太大无法预先计算导致连接在传完之前被超时断开。而且上传没有进度反馈——用户不知道已经传了多少了。痛点问题一multipart 上传 vs 流式上传——内存占用的巨大差异。files参数会把整个 multipart 请求体所有文件 所有 form 字段在内存中组装完毕后再发送。一个 5GB 的视频意味着至少 5GB 的内存峰值。而流式上传使用生成器作为data参数每次只读取一个文件块到内存、发送、释放——5GB 文件内存峰值可以控制在 8KBchunk_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 内存峰值。你需要流式上传。”小胖挠头“流式上传我学过流式下载——用streamTrueiter_content()。但上传怎么流式”大师“用生成器作为data参数。当一个生成器被用作data时requests 会自动切换到Transfer-Encoding: chunked模式——它不会一次性读取整个生成器而是 yield 一块、发一块、yield 下一块、发下一块。内存占用就降到了 chunk 大小。”deffile_chunk_generator(filepath,chunk_size8192):生成器逐块读取文件withopen(filepath,rb)asf:whilechunk:f.read(chunk_size):yieldchunk# 流式上传——内存峰值 chunk_size (8KB)requests.post(https://upload.example.com/videos,datafile_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_size8192):totalos.path.getsize(filepath)withtqdm(totaltotal,unitB,unit_scaleTrue)aspbar:withopen(filepath,rb)asf:whilechunk:f.read(chunk_size):yieldchunk pbar.update(len(chunk))小白若有所思“那如果上传中断了呢用户得从 50% 重新开始吗”大师“这就是断点续传上传——你需要服务端支持。流程是客户端先询问服务端’你已经收到了多少’GET /upload/status?file_idxxx服务端返回已接收的字节数。客户端从那个位置开始继续上传——类似于下载的Range头但上传没有标准的 Range 头通常用自定义头或 Upload-Offset。”生活比喻技术映射水管送水一点一点流流式上传生成器 chunked用卡车一次性运全部货物Content-Length 一次性上传分批快递每次送一箱chunked 分块传输快递单号实时跟踪生成器中加进度回调断掉的快递从中转站继续运断点续传上传服务端协调3. 项目实战环境准备pipinstallrequests requests-toolbelt tqdm分步实现步骤一生成器流式上传——基础版目标实现最简单的流式上传理解 chunked 编码。importrequestsimportosimporttempfiledefread_file_chunks(filepath:str,chunk_size:int65536):生成器逐块读取文件内存友好withopen(filepath,rb)asf:whileTrue:chunkf.read(chunk_size)ifnotchunk:breakyieldchunkdefstream_upload_raw(url:str,filepath:str,content_type:strapplication/octet-stream):流式上传——用生成器作为 data 参数 注意这种方式发送的是原始二进制 body不走 multipart 封装。 适合服务端接受原始二进制上传的场景如 S3 presigned URL。 file_sizeos.path.getsize(filepath)print(f文件:{os.path.basename(filepath)}({file_size:,}bytes))# 使用生成器——requests 自动启用 chunked 编码resprequests.put(url,dataread_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 会回显请求体大小tmptempfile.NamedTemporaryFile(deleteFalse,suffix.bin)tmp.write(bA*100000)# 100KB 测试文件tmp.close()respstream_upload_raw(https://httpbin.org/put,tmp.name)dataresp.json()print(f服务端收到的数据大小:{len(data.get(data,))}字节)os.unlink(tmp.name)步骤二带进度条的流式上传目标实现带实时进度、传输速度的大文件上传。importrequestsimportosimporttimefromtypingimportOptional,CallableclassProgressUploader:带进度追踪的流式上传器def__init__(self,chunk_size:int65536):self.chunk_sizechunk_sizedef_progress_generator(self,filepath:str,callback:Optional[Callable]None):生成器——逐块读取 进度回调totalos.path.getsize(filepath)uploaded0start_timetime.perf_counter()withopen(filepath,rb)asf:whilechunk:f.read(self.chunk_size):yieldchunk uploadedlen(chunk)ifcallback:elapsedtime.perf_counter()-start_time speeduploaded/elapsedifelapsed0else0pctuploaded/total*100callback(uploaded,total,speed,elapsed,pct)defupload(self,url:str,filepath:str,content_type:strapplication/octet-stream,extra_headers:dictNone,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_upuploaded/1024**2mb_totaltotal/1024**2mb_sspeed/1024**2eta(total-uploaded)/speedifspeed0else0print(f\r 进度:{mb_up:.1f}/{mb_total:.1f}MB f({pct:.0f}%){mb_s:.1f}MB/s fETA:{eta:.0f}s ,end,flushTrue)cbprogress_callbackor_default_progress resprequests.put(url,dataself._progress_generator(filepath,cb),headersheaders,timeout(10,600),)print()# 换行returnresp# 使用演示 importtempfile# 创建一个 1MB 的测试文件tmptempfile.NamedTemporaryFile(deleteFalse,suffix.bin)tmp.write(bX*(1024*1024))tmp.close()uploaderProgressUploader(chunk_size16384)respuploader.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:dictNone,progress_callbackNone,):流式 multipart 上传——支持文件 表单字段 与 requests 的 files 参数不同 - files 参数整个请求体在内存中组装 - MultipartEncoder流式读取文件逐块组装 filenameos.path.basename(filepath)file_sizeos.path.getsize(filepath)withopen(filepath,rb)asf:# 构建 multipart 字段form_fields{file:(filename,f,application/octet-stream),}iffields:form_fields.update(fields)encoderMultipartEncoder(fieldsform_fields)# 可选包装进度监控ifprogress_callback:def_progress(monitor):pctmonitor.bytes_read/monitor.len*100progress_callback(monitor.bytes_read,monitor.len,pct)monitorMultipartEncoderMonitor(encoder,_progress)datamonitorelse:dataencoder resprequests.post(url,datadata,headers{Content-Type:encoder.content_type},timeout(10,300),)returnresp# 演示 tmptempfile.NamedTemporaryFile(deleteFalse,suffix.dat)tmp.write(bHello Multipart Stream!*1000)tmp.close()defshow_progress(uploaded,total,pct):print(f\r multipart 上传:{uploaded:,}/{total:,}({pct:.0f}%),end)respstream_multipart_upload(https://httpbin.org/post,tmp.name,fields{author:小胖,department:IT},progress_callbackshow_progress,)print(f\n结果:{resp.status_code})dataresp.json()iffilesindata:print(f 服务端收到文件:{data[files].get(file,)[:50]}...)ifformindata:print(f 服务端收到字段:{data[form]})os.unlink(tmp.name)可能遇到的坑及解决方法坑1生成器只能迭代一次# 如果生成器在第一次迭代后被重用不会产生任何数据# 因为生成器已经耗尽了# 解决每次上传前创建新的生成器实例generatorread_file_chunks(file.bin)requests.put(url,datagenerator)# 第一次 OKrequests.put(url,datagenerator)# ❌ 第二次不发送任何数据坑2chunked 编码与 Content-Length 的冲突# 如果同时设置了 Content-Length 和使用生成器作为 data# urllib3 可能忽略生成器的 chunked 行为采用 Content-Length# 如果你的 Content-Length 不准确服务端会等到超时坑3流式上传不支持自动重试# 与普通请求不同生成器不能被重试因为无法重置# 如果上传失败需要从文件开头重新创建生成器# 这也就是为什么流式上传的 retry 需要客户端配合测试验证importpytestimportrequestsimporttempfileimportosclassTestStreamUpload:验证流式上传功能deftest_generator_upload_to_httpbin(self):验证用生成器流式上传tmptempfile.NamedTemporaryFile(deleteFalse,suffix.dat)tmp.write(btest data *100)tmp.close()defgen():withopen(tmp.name,rb)asf:yieldf.read()resprequests.put(https://httpbin.org/put,datagen(),)assertresp.status_code200os.unlink(tmp.name)deftest_chunked_large_data_no_oom(self):验证流式上传不会 OOMimporttracemalloc tracemalloc.start()defgen():foriinrange(100):yieldbA*8192resprequests.put(https://httpbin.org/put,datagen(),)_,peaktracemalloc.get_traced_memory()peak_mbpeak/1024**2# 内存峰值应该远小于 8KB*100 800KB实际 ~10MB 包括库开销assertresp.status_code200print(f\n 内存峰值:{peak_mb:.1f}MB)tracemalloc.stop()deftest_content_known_size_with_length(self):已知大小时传 Content-Lengthdatabhello world*100resprequests.put(https://httpbin.org/put,datadata,headers{Content-Length:str(len(data))},)assertresp.status_code2004. 项目总结核心知识点上传方式内存占用适用场景Content-Typefiles{}高文件大小小文件 表单multipart/form-datadatabytes高文件大小小文件纯二进制自定义datagenerator低chunk_size大文件流式chunkedMultipartEncoder低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请求头。你需要设计一个客户端-服务端协议来支持这个功能。延伸阅读与资源Milvus向量数据库实战修炼从 0 到 1精通向量检索与生产落地后端工程师的 AI 转型第一课Ollama 与私有化大模型实战10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析
)
