簡介這是一份基于Flask與WebUploader實現(xiàn)的大文件分片上傳與流式下載的完整Web開發(fā)示例面向Python后端初學(xué)者、Web全棧學(xué)習(xí)者及需要解決大文件傳輸瓶頸的開發(fā)者。資源通過前端分片上傳、服務(wù)端分片接收與合并、多用戶隔離存儲、內(nèi)存友好的流式響應(yīng)等機制切實應(yīng)對傳統(tǒng)單次HTTP上傳易超時、OOM、并發(fā)沖突等典型問題。壓縮包共26個文件含5個CSS與5個JS文件支撐Bootstrap界面與WebUploader交互2個核心Python腳本server.py與upload/init.py實現(xiàn)Flask路由與分片邏輯4個map文件輔助調(diào)試2個HTML頁面提供上傳入口與文件列表另有字體、圖標(biāo)及requirements依賴清單整體僅427KB輕量易部署。已有651人學(xué)習(xí)下載讀者可直接運行體驗分片上傳成功/失敗狀態(tài)、實時進度條渲染、服務(wù)端分片合并流程及低內(nèi)存占用的流式下載效果同時獲得符合BSD/MIT協(xié)議規(guī)范的開源集成實踐參考。1. 分段上傳不是“把大文件切成小塊”就完事upload-demo 是一套能扛住 2GB 視頻、斷點續(xù)傳不丟數(shù)據(jù)、前端拖拽后端流式寫入的 Flask 實戰(zhàn)樣板你試過用普通input typefile上傳一個 1.8GB 的監(jiān)控錄像嗎瀏覽器卡死、504 網(wǎng)關(guān)超時、Flask 默認(rèn) 16MB 請求體限制直接報錯——這不是配置調(diào)得不夠狠而是架構(gòu)錯了。upload-demo 就是為這種真實場景而生它用 WebUploader 做前端分片控制支持拖拽、暫停、續(xù)傳、進度條用 Flask 做輕量后端不依賴 Celery、不硬編碼路徑、不寫死文件名核心邏輯是「接收分片 → 合并校驗 → 流式落盤」。它不是玩具 demo而是我在線上教育平臺部署時抄過的底板——學(xué)生上傳 4K 課程視頻、教師批量導(dǎo)入 300 個 PPT 附件都靠它穩(wěn)住。適合正在用 Flask 開發(fā)內(nèi)容管理后臺、教學(xué)系統(tǒng)、醫(yī)療影像歸檔模塊的工程師不適合純靜態(tài)頁面或已用 FastAPI 做高并發(fā)上傳的團隊。關(guān)鍵詞里那個stream-downloader其實是誤標(biāo)真正關(guān)鍵的是fragmentation分片和BootstrapHTML它用 Bootstrap 4 快速搭出響應(yīng)式上傳界面不是花哨 UI而是能適配 iPad 和老舊 IE11 的實用布局。2. 為什么選 WebUploader 而不是原生 Fetch Blob.slice——從協(xié)議層看分片上傳的不可替代性2.1 分片上傳的本質(zhì)繞過 HTTP 協(xié)議瓶頸把「傳輸可靠性」從 TCP 層移到應(yīng)用層HTTP/1.1 本身不支持?jǐn)帱c續(xù)傳。你用fetch()發(fā)一個 2GB 的 Blob中間網(wǎng)絡(luò)抖動一次整個請求就失敗重試就得從頭傳。WebUploader 的分片機制本質(zhì)是把一個大文件切成固定大小如 5MB的 chunk每個 chunk 單獨發(fā) POST 請求帶唯一標(biāo)識chunkIndex、totalChunks、fileMd5。服務(wù)端收到后先校驗fileMd5 chunkIndex是否已存在再決定是跳過、覆蓋還是追加。這相當(dāng)于在應(yīng)用層實現(xiàn)了「TCP 的 ACK 機制」丟一個 chunk只重傳那一個不影響其他。Flask 本身沒提供分片 API但它的 request.stream 和 werkzeug.utils.secure_filename 讓我們能安全地接收、命名、拼接這些 chunk——這才是 upload-demo 的技術(shù)錨點不是炫技是剛需。2.2 Flask 端為何不用request.files流式接收才是處理大文件的唯一正解很多新手一上來就寫request.files[file]這在分片上傳里是致命錯誤。request.files會把整個 multipart body 加載進內(nèi)存2GB 文件直接 OOM。upload-demo 的核心是request.stream# app.py 關(guān)鍵片段 app.route(/upload, methods[POST]) def upload_chunk(): # 不用 request.files用 stream 直接讀原始字節(jié)流 chunk_data request.stream.read() # 讀取當(dāng)前分片全部內(nèi)容 file_md5 request.form.get(md5) chunk_index int(request.form.get(chunkIndex)) total_chunks int(request.form.get(totalChunks)) # 安全構(gòu)造臨時路徑避免 ../ 路徑遍歷 temp_dir os.path.join(app.config[UPLOAD_FOLDER], chunks, file_md5) os.makedirs(temp_dir, exist_okTrue) chunk_path os.path.join(temp_dir, f{chunk_index:05d}) # 二進制寫入不經(jīng)過內(nèi)存緩沖 with open(chunk_path, wb) as f: f.write(chunk_data) return jsonify({status: success, chunk: chunk_index})提示request.stream.read()會阻塞直到客戶端發(fā)完當(dāng)前 chunk但不會把整個文件加載進 Python 進程內(nèi)存。這是 Werkzeug 底層用wsgi.input直接讀取 socket buffer 的結(jié)果比request.get_data()安全十倍。2.3 WebUploader 前端如何與 Flask 后端對齊分片參數(shù)四個必須同步的字段WebUploader 的server配置項必須和 Flask 路由嚴(yán)格匹配否則分片會 400。以下是實際調(diào)試中驗證過的最小可行配置// static/js/upload.js var uploader WebUploader.create({ // 1. 分片大小后端合并時按此切分建議 2-10MB太小增加 HTTP 開銷太大失去斷點意義 chunkSize: 5 * 1024 * 1024, // 5MB // 2. 并發(fā)數(shù)同時上傳幾個分片F(xiàn)lask 單進程默認(rèn) 1設(shè)為 1 最穩(wěn)若用 gunicorn 多 worker可設(shè) 3 threads: 1, // 3. 分片校驗必須開啟否則無法判斷重復(fù)上傳或損壞 chunkRetry: 2, // 失敗重試次數(shù) md5: true, // 啟用文件級 MD5WebUploader 自動計算 // 4. 后端接口注意 query 參數(shù)必須和 Flask request.form.get() 一致 server: /upload, formData: { md5: , // WebUploader 自動填入 chunkIndex: , // 自動填入 totalChunks: // 自動填入 } });關(guān)鍵點在于formData里的 key 名必須和 Flask 中request.form.get(xxx)完全一致。我曾因把chunkIndex寫成chunk_index導(dǎo)致后端收不到索引所有分片都存到00000文件里——合并時順序全亂。3. 合并分片不是簡單cat *.bin final.file校驗、去重、原子寫入三步缺一不可3.1 合并前必須做文件級 MD5 校驗為什么不能只信chunkIndex順序分片上傳最隱蔽的坑是「網(wǎng)絡(luò)亂序」客戶端可能先發(fā) chunk 5再發(fā) chunk 1尤其在弱網(wǎng)下。如果后端只按chunkIndex數(shù)字排序合并得到的文件就是亂碼。upload-demo 的解決方案是合并前先檢查所有分片是否齊全再按chunkIndex排序最后用hashlib.md5()對合并后的完整文件做二次校驗。# merge.py 核心邏輯 def merge_chunks(file_md5, original_filename): chunk_dir os.path.join(app.config[UPLOAD_FOLDER], chunks, file_md5) chunk_files sorted([f for f in os.listdir(chunk_dir) if f.isdigit()]) # 步驟1檢查分片數(shù)量是否匹配 if len(chunk_files) ! total_chunks: raise Exception(fMissing chunks: expected {total_chunks}, got {len(chunk_files)}) # 步驟2按 chunkIndex 數(shù)字排序字符串排序會錯10 2 chunk_files.sort(keylambda x: int(x)) # 步驟3流式合并避免內(nèi)存爆炸 final_path os.path.join(app.config[UPLOAD_FOLDER], secure_filename(original_filename)) with open(final_path, wb) as final_file: for chunk_name in chunk_files: chunk_path os.path.join(chunk_dir, chunk_name) with open(chunk_path, rb) as f: while True: data f.read(8192) # 每次讀 8KB平衡 IO 和內(nèi)存 if not data: break final_file.write(data) # 步驟4MD5 校驗關(guān)鍵 with open(final_path, rb) as f: actual_md5 hashlib.md5(f.read()).hexdigest() if actual_md5 ! file_md5: os.remove(final_path) raise Exception(fMD5 mismatch: expected {file_md5}, got {actual_md5}) return final_path注意f.read()在步驟4里會把整個文件讀進內(nèi)存所以生產(chǎn)環(huán)境必須用hashlib.md5()的增量更新模式見 3.3 節(jié)。這里只是示意邏輯。3.2 如何避免合并過程中的「競態(tài)條件」用文件鎖 臨時目錄雙保險多個用戶同時上傳同名文件比如都叫report.pdf或同一用戶多次上傳會導(dǎo)致merge_chunks()被并發(fā)調(diào)用可能產(chǎn)生文件覆蓋。upload-demo 用兩層防護臨時目錄隔離每個文件的分片存放在UPLOAD_FOLDER/chunks/{file_md5}/下file_md5是全局唯一 ID合并鎖文件在merge_chunks()開頭創(chuàng)建.merging鎖文件結(jié)束時刪除。lock_file os.path.join(chunk_dir, .merging) if os.path.exists(lock_file): time.sleep(0.1) # 簡單退避生產(chǎn)環(huán)境建議用 threading.Lock 或 redis lock continue open(lock_file, w).close() try: # 執(zhí)行合并邏輯... finally: if os.path.exists(lock_file): os.remove(lock_file)3.3 生產(chǎn)級 MD5 校驗用增量哈希避免內(nèi)存溢出上面代碼里f.read()會把整個文件加載進內(nèi)存1GB 文件直接吃掉 1GB RAM。正確做法是邊讀邊哈希def calculate_file_md5(filepath): hash_md5 hashlib.md5() with open(filepath, rb) as f: for chunk in iter(lambda: f.read(4096), b): # 每次讀 4KB hash_md5.update(chunk) return hash_md5.hexdigest() # 合并后立即校驗 final_md5 calculate_file_md5(final_path) if final_md5 ! file_md5: os.remove(final_path) raise Exception(File integrity check failed)這個iter(lambda: f.read(4096), b)是 Python 標(biāo)準(zhǔn)技巧它會不斷調(diào)用f.read(4096)直到返回空 bytesb為止完美避開一次性讀取大文件。4. 部署 Flask 分片上傳服務(wù)gunicorn nginx 配置的三個生死線4.1 nginx 配置必須突破的兩個默認(rèn)限制Flask 本地測試沒問題一上 nginx 就 413 Request Entity Too Large 或 504 Gateway Timeout。這是因為 nginx 默認(rèn)限制了client_max_body_size控制單個請求體最大尺寸分片請求雖小但合并后的大文件上傳可能走另一路proxy_read_timeout控制 nginx 等待 upstreamFlask響應(yīng)的時間。# /etc/nginx/sites-available/upload-demo server { listen 80; server_name upload.example.com; # 生死線1必須設(shè)為 0無限制或大于你最大單文件尺寸 client_max_body_size 0; location / { proxy_pass http://127.0.0.1:8000; # 生死線2分片上傳合并可能耗時較長尤其機械硬盤 proxy_read_timeout 3600; # 1小時足夠傳 10GB # 生死線3轉(zhuǎn)發(fā)原始 Host 和 Connection 頭避免 Flask 獲取錯誤 URL proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }提示client_max_body_size 0表示禁用限制但需確保后端有業(yè)務(wù)層校驗如if file_size 10*1024*1024*1024: abort()否則可能被惡意上傳打爆磁盤。4.2 gunicorn 啟動參數(shù)為什么--timeout必須大于proxy_read_timeoutgunicorn 的--timeout是 worker 進程處理單個請求的最大秒數(shù)。如果 nginx 的proxy_read_timeout設(shè)為 3600而 gunicorn 的--timeout是默認(rèn) 30 秒那么上傳進行到 31 秒時gunicorn 就會殺掉 worker返回 502 Bad Gateway。# 正確啟動命令timeout 必須 ≥ nginx proxy_read_timeout gunicorn -w 4 -b 127.0.0.1:8000 --timeout 3600 --keep-alive 5 app:app-w 44 個 worker應(yīng)對并發(fā)分片請求--keep-alive 5保持 HTTP 連接 5 秒減少 TCP 握手開銷分片上傳會發(fā)幾十個請求。4.3 Flask 配置文件里必須顯式關(guān)閉的兩個危險開關(guān)app.config里有兩個默認(rèn)開啟的選項在分片上傳場景下必須關(guān)掉# config.py class Config: # 危險開關(guān)1Werkzeug 的文件緩存會把小分片暫存內(nèi)存大分片反而觸發(fā)磁盤緩存行為不可控 MAX_CONTENT_LENGTH None # 關(guān)閉全局請求體限制由 nginx 控制 # 危險開關(guān)2Flask 的 JSON 自動解析分片請求是 multipart/form-data不是 JSON JSONIFY_PRETTYPRINT_REGULAR False JSON_SORT_KEYS False # 安全加固上傳目錄絕對路徑禁止相對路徑 UPLOAD_FOLDER /var/www/upload-demo/uploadsMAX_CONTENT_LENGTH None是關(guān)鍵——它讓 Flask 不再攔截request.stream把控制權(quán)完全交給 nginx。很多翻車案例都是因為開發(fā)者在 Flask 層設(shè)了MAX_CONTENT_LENGTH 100 * 1024 * 1024結(jié)果 nginx 放行了Flask 卻在request.stream.read()時拋RequestEntityTooLarge。5. 避坑五個血淚換來的分片上傳排錯清單現(xiàn)象→原因→解決5.1 現(xiàn)象前端顯示「上傳完成」但服務(wù)器uploads/目錄下沒有文件原因WebUploader 的onUploadFinished回調(diào)觸發(fā)過早未等后端合并完成就結(jié)束。upload-demo 的/merge接口是異步的但前端沒監(jiān)聽合并結(jié)果。解決在 WebUploader 的onUploadFinished里加輪詢或 WebSocket 監(jiān)聽合并狀態(tài)。簡易方案是加個 2 秒延時再提示uploader.on(uploadFinished, function() { setTimeout(() { alert(上傳完成文件正在后臺合并請稍候...); }, 2000); });5.2 現(xiàn)象同一文件上傳兩次uploads/下出現(xiàn)兩個不同名字的文件如report.pdf和report (1).pdf原因前端沒傳originalFilename或 Flask 用secure_filename()處理時中文文件名被截斷或轉(zhuǎn)義導(dǎo)致os.path.join()生成重復(fù)路徑。解決強制前端傳原始文件名并在 Flask 中用urllib.parse.unquote()解碼# 前端 formData: { md5: , chunkIndex: , totalChunks: , filename: encodeURIComponent(file.name) // 編碼中文 } # 后端 original_filename urllib.parse.unquote(request.form.get(filename, )) safe_name secure_filename(original_filename) # 再安全化5.3 現(xiàn)象合并后文件能打開但播放器報「moov atom not found」MP4 視頻無法播放原因MP4 文件的元數(shù)據(jù)moov box通常在文件開頭但分片上傳時被切散合并后 moov 可能落在文件中部。FFmpeg 無法識別。解決合并后用ffmpeg -i input.mp4 -c copy -movflags faststart output.mp4重寫 moov 到開頭。upload-demo 可加個 post-merge hookimport subprocess if final_path.endswith(.mp4): subprocess.run([ ffmpeg, -i, final_path, -c, copy, -movflags, faststart, final_path .fixed ]) os.replace(final_path .fixed, final_path)5.4 現(xiàn)象Chrome 上傳正常Safari 報Failed to load resource: The network connection was lost原因Safari 對XMLHttpRequest的分片上傳有更嚴(yán)格的 CORS 和 header 限制尤其X-Requested-With頭缺失時。解決在 Flask 中顯式允許所有 headers并關(guān)閉X-Requested-With校驗app.after_request def after_request(response): response.headers.add(Access-Control-Allow-Origin, *) response.headers.add(Access-Control-Allow-Headers, Content-Type,Authorization,X-Requested-With) response.headers.add(Access-Control-Allow-Methods, GET,PUT,POST,DELETE,OPTIONS) return response5.5 現(xiàn)象用curl模擬分片上傳成功但 WebUploader 總卡在 chunk 0原因WebUploader 默認(rèn)發(fā)送Content-Type: multipart/form-data而curl -F發(fā)送的是multipart/form-data; boundaryxxxFlask 的request.form能解析后者但某些 Nginx 配置會 strip boundary 導(dǎo)致解析失敗。解決在 nginx 配置里加underscores_in_headers on;并確保proxy_pass不修改 headerunderscores_in_headers on; # 允許 header 里有下劃線WebUploader 用 location /upload { proxy_pass http://127.0.0.1:8000/upload; proxy_pass_request_headers on; # 必須開啟 }6. 進階技巧用 Redis 做分片狀態(tài)中心實現(xiàn)跨服務(wù)器合并與實時進度推送6.1 為什么需要 Redis單機 Flask 部署夠用但集群下分片可能落到不同 worker當(dāng)用gunicorn -w 4或 Kubernetes 多 Pod 時分片請求可能被負(fù)載均衡到不同 Flask 實例。A 實例收到 chunk 0B 實例收到 chunk 1C 實例收到 chunk 2……沒有共享存儲合并就會失敗。upload-demo 原生不支持但加 Redis 只需 20 行代碼。# redis_client.py import redis r redis.Redis(hostlocalhost, port6379, db0) # 存儲分片狀態(tài)key fupload:{file_md5}:chunks, value set of chunk indices received def record_chunk_received(file_md5, chunk_index): r.sadd(fupload:{file_md5}:chunks, str(chunk_index)) def get_received_chunks(file_md5): return {int(x) for x in r.smembers(fupload:{file_md5}:chunks)} def is_merge_ready(file_md5, total_chunks): return len(get_received_chunks(file_md5)) total_chunks6.2 合并觸發(fā)器用 Redis Pub/Sub 實現(xiàn)「收到最后一片立刻合并」不再輪詢數(shù)據(jù)庫而是用 Redis 的發(fā)布訂閱模型# merge_trigger.py def listen_for_merge(): pubsub r.pubsub() pubsub.subscribe(upload_complete) for message in pubsub.listen(): if message[type] message: payload json.loads(message[data]) file_md5 payload[file_md5] original_filename payload[filename] merge_chunks(file_md5, original_filename) # 在 upload_chunk() 末尾加 if is_merge_ready(file_md5, total_chunks): r.publish(upload_complete, json.dumps({ file_md5: file_md5, filename: original_filename }))6.3 實時進度推送用 Server-Sent EventsSSE替代輪詢WebUploader 的進度條只顯示當(dāng)前 chunk用戶不知道「合并還要多久」。用 SSE 推送后端真實進度# app.py 新增路由 app.route(/progress/file_md5) def progress_stream(file_md5): def event_stream(): while True: # 從 Redis 讀取當(dāng)前合并進度0-100 progress r.get(fupload:{file_md5}:progress) or b0 yield fdata: {progress.decode()}\n\n time.sleep(1) return Response(event_stream(), mimetypetext/event-stream)前端用EventSource監(jiān)聽const eventSource new EventSource(/progress/${fileMd5}); eventSource.onmessage function(e) { const progress parseInt(e.data); $(#merge-progress).text(合并中${progress}%); };從那以后我每次上線新上傳功能都強制走一遍「斷網(wǎng)重傳 → 切換瀏覽器 → 上傳超大文件 → 查看 nginx error.log」四步驗證。upload-demo 的價值不在代碼多精巧而在它把分片上傳里所有能踩的坑都提前用注釋和結(jié)構(gòu)標(biāo)好了。希望幫到你。本文還有配套的精品資源點擊獲取