前言在线PDF翻译服务通常限制单文件大小如20MB但对于开发者来说集成翻译功能时还面临另一个挑战如何在服务端实现大文件的可靠上传。本文以PDF翻译场景为例深入探讨流式上传的三种核心技术方案分片上传、断点续传和实时进度反馈。这三种技术不仅适用于PDF翻译场景任何需要上传大文件到云服务的应用都可以复用这些方案。为什么需要流式上传传统的文件上传方式是整块上传——客户端一次性将整个文件发送到服务端。这种方式有以下问题内存压力大服务端需要将整个文件加载到内存再处理网络不稳定导致失败上传到99%断网需要从头开始无法显示进度用户不知道要等多久超时风险高大文件如100页PDF50MB可能在上传过程中超时流式上传的核心思想是分而治之——将大文件切分为小块逐块上传最终在服务端拼接还原。方案一分片上传原理客户端将文件切分为固定大小的分片如5MB/片逐片上传到服务端。服务端接收所有分片后按顺序拼接。客户端实现PythonimportosimporthashlibimportrequestsfromtypingimportOptional,CallableclassChunkedUploader:分片上传器def__init__(self,upload_url:str,chunk_size:int5*1024*1024,# 5MBmax_retries:int3,):self.upload_urlupload_url self.chunk_sizechunk_size self.max_retriesmax_retriesdefupload(self,file_path:str,progress_callback:Optional[Callable[[int,int],None]]None,)-dict:分片上传文件 Args: file_path: 文件路径 progress_callback: 进度回调 (已上传字节, 总字节) Returns: API响应 file_sizeos.path.getsize(file_path)file_hashself._compute_file_hash(file_path)total_chunks(file_sizeself.chunk_size-1)//self.chunk_size# Step 1: 初始化上传会话session_idself._init_session(file_path,file_size,total_chunks,file_hash)# Step 2: 逐片上传withopen(file_path,rb)asf:forchunk_idxinrange(total_chunks):offsetchunk_idx*self.chunk_size f.seek(offset)chunk_dataf.read(self.chunk_size)# 计算分片哈希用于校验chunk_hashhashlib.md5(chunk_data).hexdigest()# 上传分片带重试self._upload_chunk(session_id,chunk_idx,chunk_data,chunk_hash)# 回调进度ifprogress_callback:uploadedoffsetlen(chunk_data)progress_callback(uploaded,file_size)# Step 3: 通知服务端合并分片resultself._complete_session(session_id)returnresultdef_compute_file_hash(self,file_path:str,algorithm:strmd5)-str:计算文件哈希用于去重和完整性校验hhashlib.new(algorithm)withopen(file_path,rb)asf:whilechunk:f.read(8192):h.update(chunk)returnh.hexdigest()def_init_session(self,file_path:str,file_size:int,total_chunks:int,file_hash:str)-str:初始化上传会话resprequests.post(f{self.upload_url}/session,json{file_name:os.path.basename(file_path),file_size:file_size,total_chunks:total_chunks,chunk_size:self.chunk_size,file_hash:file_hash,},timeout30,)resp.raise_for_status()returnresp.json()[session_id]def_upload_chunk(self,session_id:str,chunk_idx:int,data:bytes,chunk_hash:str):上传单个分片带指数退避重试forattemptinrange(self.max_retries):try:resprequests.put(f{self.upload_url}/session/{session_id}/chunk/{chunk_idx},datadata,headers{Content-Type:application/octet-stream,X-Chunk-Hash:chunk_hash,X-Chunk-Index:str(chunk_idx),},timeout60,)resp.raise_for_status()return# 上传成功exceptrequests.RequestExceptionase:ifattemptself.max_retries-1:raisewait2**attempt# 指数退避: 1s, 2s, 4simporttime time.sleep(wait)def_complete_session(self,session_id:str)-dict:通知服务端合并分片resprequests.post(f{self.upload_url}/session/{session_id}/complete,timeout120,# 合并可能需要较长时间)resp.raise_for_status()returnresp.json()服务端实现要点伪代码# 服务端需要维护上传会话状态# 推荐使用Redis存储会话信息session_store{}# 生产环境使用Redis# POST /session - 创建上传会话defcreate_session(file_name,file_size,total_chunks,chunk_size,file_hash):session_idgenerate_uuid()session_store[session_id]{file_name:file_name,file_size:file_size,total_chunks:total_chunks,chunk_size:chunk_size,file_hash:file_hash,received_chunks:set(),chunk_dir:f/tmp/uploads/{session_id}/,}os.makedirs(session_store[session_id][chunk_dir])return{session_id:session_id}# PUT /session/{id}/chunk/{idx} - 接收分片defreceive_chunk(session_id,chunk_idx,chunk_data,chunk_hash):# 校验分片哈希ifhashlib.md5(chunk_data).hexdigest()!chunk_hash:raiseHTTPException(400,Chunk hash mismatch)# 保存分片chunk_pathf{chunk_dir}/{chunk_idx:06d}withopen(chunk_path,wb)asf:f.write(chunk_data)session_store[session_id][received_chunks].add(chunk_idx)return{received:len(session_store[session_id][received_chunks])}方案二断点续传分片上传的一个直接扩展就是断点续传——当上传中断时可以从上次中断的位置继续而不是从头开始。核心实现classResumableUploader(ChunkedUploader):支持断点续传的上传器def_get_uploaded_chunks(self,session_id:str)-set:从服务端获取已上传的分片列表resprequests.get(f{self.upload_url}/session/{session_id}/status,timeout10,)resp.raise_for_status()statusresp.json()returnset(status.get(received_chunks,[]))defupload(self,file_path:str,session_id:strNone,**kwargs):支持断点续传的上传方法file_sizeos.path.getsize(file_path)total_chunks(file_sizeself.chunk_size-1)//self.chunk_sizeifsession_id:# 恢复已有会话uploaded_chunksself._get_uploaded_chunks(session_id)print(f恢复上传:{len(uploaded_chunks)}/{total_chunks}已完成)else:# 新建会话file_hashself._compute_file_hash(file_path)session_idself._init_session(file_path,file_size,total_chunks,file_hash)uploaded_chunksset()# 只上传未完成的分片withopen(file_path,rb)asf:forchunk_idxinrange(total_chunks):ifchunk_idxinuploaded_chunks:continue# 跳过已上传的分片offsetchunk_idx*self.chunk_size f.seek(offset)chunk_dataf.read(self.chunk_size)chunk_hashhashlib.md5(chunk_data).hexdigest()self._upload_chunk(session_id,chunk_idx,chunk_data,chunk_hash)returnself._complete_session(session_id)方案三实时进度推送WebSocket Server-Sent Events上传过程中客户端需要实时展示进度。两种主流方案SSEServer-Sent Events实现适合服务端→客户端单向推送场景# 服务端FastAPIfromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponseimportasyncioimportjson appFastAPI()app.get(/upload/session/{session_id}/progress)asyncdefupload_progress(session_id:str):通过SSE推送上传进度asyncdefevent_stream():whileTrue:# 从Redis获取当前进度progressawaitget_session_progress(session_id)yieldfdata:{json.dumps(progress)}\n\nifprogress[status]in(completed,failed):breakawaitasyncio.sleep(1)# 每秒推送一次returnStreamingResponse(event_stream(),media_typetext/event-stream,headers{Cache-Control:no-cache,X-Accel-Buffering:no,# 禁用Nginx缓冲})// 客户端JavaScript/ReactfunctionuseUploadProgress(sessionId){const[progress,setProgress]useState({percent:0,status:uploading});useEffect((){consteventSourcenewEventSource(/upload/session/${sessionId}/progress);eventSource.onmessage(event){constdataJSON.parse(event.data);setProgress(data);if(data.statuscompleted){eventSource.close();}};eventSource.onerror(){eventSource.close();setProgress(prev({...prev,status:error}));};return()eventSource.close();},[sessionId]);returnprogress;}实际应用PDF翻译上传流程整合将以上方案整合到PDF翻译场景中classPDFTranslationUploader:PDF翻译专用上传器分片上传 进度回调 翻译轮询def__init__(self,api_base_url:str):self.uploaderResumableUploader(f{api_base_url}/upload)self.api_base_urlapi_base_urldeftranslate_pdf(self,pdf_path:str,source_lang:strauto,target_lang:strzh,on_upload_progress:Optional[Callable]None,on_translation_progress:Optional[Callable]None,)-dict:完整的PDF翻译流程# Phase 1: 流式上传print(f 开始上传:{os.path.basename(pdf_path)})upload_resultself.uploader.upload(pdf_path,progress_callbackon_upload_progress,)print(f✅ 上传完成:{upload_result[file_id]})# Phase 2: 提交翻译任务task_idself._submit_translation(upload_result[file_id],source_lang,target_lang,)print(f 翻译中... task_id{task_id})# Phase 3: 轮询翻译进度resultself._poll_translation(task_id,on_translation_progress)print(f✅ 翻译完成)returnresultdef_submit_translation(self,file_id:str,source_lang:str,target_lang:str)-str:resprequests.post(f{self.api_base_url}/translate,json{file_id:file_id,source_lang:source_lang,target_lang:target_lang,},timeout30,)resp.raise_for_status()returnresp.json()[task_id]def_poll_translation(self,task_id:str,progress_callbackNone,max_wait:int600)-dict:轮询翻译状态最多等待10分钟importtime starttime.time()whiletime.time()-startmax_wait:resprequests.get(f{self.api_base_url}/task/{task_id},timeout10,)dataresp.json()ifprogress_callback:progress_callback(data.get(progress,0))ifdata[status]completed:returndataelifdata[status]failed:raiseException(f翻译失败:{data.get(error)})time.sleep(2)# 每2秒查询一次raiseTimeoutError(翻译超时)性能对比方案上传耗时100MB文件内存占用断点续传进度显示整块上传45s100MB❌❌流式上传12s5MB❌分片级流式分片14s5MB✅分片级流式分片并发6s15MB✅分片级测试环境100Mbps宽带5个并发分片单分片5MB总结流式上传不是一个新技术但在PDF翻译这类SaaS产品的工程实现中它直接决定了用户体验的好坏。三个核心要点分片上传解决了大文件传输的可靠性问题断点续传在网络不稳定场景下大幅提升成功率实时进度反馈SSE/WebSocket让等待不再是黑盒对于开发者来说建议先从分片上传开始实现再逐步加入断点续传和进度推送。完整的实现代码可在GitHub上找到也可以直接参考成熟的云服务SDK如AWS S3的分片上传API来理解最佳实践。标签PDF翻译、文件上传、Python实战、性能优化、SaaS