簡介這是一份基于Flask與WebUploader實(shí)現(xiàn)的大文件分片上傳與流式下載的完整Web開發(fā)示例面向Python后端初學(xué)者、Web全棧學(xué)習(xí)者及需要解決大文件傳輸瓶頸的開發(fā)者。資源通過前端分片上傳、服務(wù)端分片接收與合并、多用戶隔離存儲(chǔ)、內(nèi)存友好的流式響應(yīng)等機(jī)制切實(shí)應(yīng)對傳統(tǒng)單次HTTP上傳易超時(shí)、OOM、并發(fā)沖突等典型問題。壓縮包共26個(gè)文件含5個(gè)CSS與5個(gè)JS文件支撐Bootstrap界面與WebUploader交互2個(gè)核心Python腳本server.py與upload/init.py實(shí)現(xiàn)Flask路由與分片邏輯4個(gè)map文件輔助調(diào)試2個(gè)HTML頁面提供上傳入口與文件列表另有字體、圖標(biāo)及requirements依賴清單整體僅427KB輕量易部署。已有651人學(xué)習(xí)下載讀者可直接運(yùn)行體驗(yàn)分片上傳成功/失敗狀態(tài)、實(shí)時(shí)進(jìn)度條渲染、服務(wù)端分片合并流程及低內(nèi)存占用的流式下載效果同時(shí)獲得符合BSD/MIT協(xié)議規(guī)范的開源集成實(shí)踐參考。1. 分段上傳不是“把大文件切成小塊”就完事upload-demo 是一套能扛住 2GB 視頻、斷點(diǎn)續(xù)傳不丟數(shù)據(jù)、前端拖拽后端流式寫入的 Flask 實(shí)戰(zhàn)樣板你試過用普通input typefile上傳一個(gè) 1.8GB 的監(jiān)控錄像嗎瀏覽器卡死、504 網(wǎng)關(guān)超時(shí)、Flask 默認(rèn) 16MB 請求體限制直接報(bào)錯(cuò)——這不是配置調(diào)得不夠狠而是架構(gòu)錯(cuò)了。upload-demo 就是為這種真實(shí)場景而生它用 WebUploader 做前端分片控制支持拖拽、暫停、續(xù)傳、進(jìn)度條用 Flask 做輕量后端不依賴 Celery、不硬編碼路徑、不寫死文件名核心邏輯是「接收分片 → 合并校驗(yàn) → 流式落盤」。它不是玩具 demo而是我在線上教育平臺(tái)部署時(shí)抄過的底板——學(xué)生上傳 4K 課程視頻、教師批量導(dǎo)入 300 個(gè) PPT 附件都靠它穩(wěn)住。適合正在用 Flask 開發(fā)內(nèi)容管理后臺(tái)、教學(xué)系統(tǒng)、醫(yī)療影像歸檔模塊的工程師不適合純靜態(tài)頁面或已用 FastAPI 做高并發(fā)上傳的團(tuán)隊(duì)。關(guān)鍵詞里那個(gè)stream-downloader其實(shí)是誤標(biāo)真正關(guān)鍵的是fragmentation分片和BootstrapHTML它用 Bootstrap 4 快速搭出響應(yīng)式上傳界面不是花哨 UI而是能適配 iPad 和老舊 IE11 的實(shí)用布局。2. 為什么選 WebUploader 而不是原生 Fetch Blob.slice——從協(xié)議層看分片上傳的不可替代性2.1 分片上傳的本質(zhì)繞過 HTTP 協(xié)議瓶頸把「傳輸可靠性」從 TCP 層移到應(yīng)用層HTTP/1.1 本身不支持?jǐn)帱c(diǎn)續(xù)傳。你用fetch()發(fā)一個(gè) 2GB 的 Blob中間網(wǎng)絡(luò)抖動(dòng)一次整個(gè)請求就失敗重試就得從頭傳。WebUploader 的分片機(jī)制本質(zhì)是把一個(gè)大文件切成固定大小如 5MB的 chunk每個(gè) chunk 單獨(dú)發(fā) POST 請求帶唯一標(biāo)識(shí)chunkIndex、totalChunks、fileMd5。服務(wù)端收到后先校驗(yàn)fileMd5 chunkIndex是否已存在再?zèng)Q定是跳過、覆蓋還是追加。這相當(dāng)于在應(yīng)用層實(shí)現(xiàn)了「TCP 的 ACK 機(jī)制」丟一個(gè) chunk只重傳那一個(gè)不影響其他。Flask 本身沒提供分片 API但它的 request.stream 和 werkzeug.utils.secure_filename 讓我們能安全地接收、命名、拼接這些 chunk——這才是 upload-demo 的技術(shù)錨點(diǎn)不是炫技是剛需。2.2 Flask 端為何不用request.files流式接收才是處理大文件的唯一正解很多新手一上來就寫request.files[file]這在分片上傳里是致命錯(cuò)誤。request.files會(huì)把整個(gè) multipart body 加載進(jìn)內(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)造臨時(shí)路徑避免 ../ 路徑遍歷 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ìn)制寫入不經(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()會(huì)阻塞直到客戶端發(fā)完當(dāng)前 chunk但不會(huì)把整個(gè)文件加載進(jìn) Python 進(jìn)程內(nèi)存。這是 Werkzeug 底層用wsgi.input直接讀取 socket buffer 的結(jié)果比request.get_data()安全十倍。2.3 WebUploader 前端如何與 Flask 后端對齊分片參數(shù)四個(gè)必須同步的字段WebUploader 的server配置項(xiàng)必須和 Flask 路由嚴(yán)格匹配否則分片會(huì) 400。以下是實(shí)際調(diào)試中驗(yàn)證過的最小可行配置// static/js/upload.js var uploader WebUploader.create({ // 1. 分片大小后端合并時(shí)按此切分建議 2-10MB太小增加 HTTP 開銷太大失去斷點(diǎn)意義 chunkSize: 5 * 1024 * 1024, // 5MB // 2. 并發(fā)數(shù)同時(shí)上傳幾個(gè)分片F(xiàn)lask 單進(jìn)程默認(rèn) 1設(shè)為 1 最穩(wěn)若用 gunicorn 多 worker可設(shè) 3 threads: 1, // 3. 分片校驗(yàn)必須開啟否則無法判斷重復(fù)上傳或損壞 chunkRetry: 2, // 失敗重試次數(shù) md5: true, // 啟用文件級(jí) MD5WebUploader 自動(dòng)計(jì)算 // 4. 后端接口注意 query 參數(shù)必須和 Flask request.form.get() 一致 server: /upload, formData: { md5: , // WebUploader 自動(dòng)填入 chunkIndex: , // 自動(dòng)填入 totalChunks: // 自動(dòng)填入 } });關(guān)鍵點(diǎn)在于formData里的 key 名必須和 Flask 中request.form.get(xxx)完全一致。我曾因把chunkIndex寫成chunk_index導(dǎo)致后端收不到索引所有分片都存到00000文件里——合并時(shí)順序全亂。3. 合并分片不是簡單cat *.bin final.file校驗(yàn)、去重、原子寫入三步缺一不可3.1 合并前必須做文件級(jí) MD5 校驗(yàn)為什么不能只信chunkIndex順序分片上傳最隱蔽的坑是「網(wǎng)絡(luò)亂序」客戶端可能先發(fā) chunk 5再發(fā) chunk 1尤其在弱網(wǎng)下。如果后端只按chunkIndex數(shù)字排序合并得到的文件就是亂碼。upload-demo 的解決方案是合并前先檢查所有分片是否齊全再按chunkIndex排序最后用hashlib.md5()對合并后的完整文件做二次校驗(yàn)。# 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ù)字排序字符串排序會(huì)錯(cuò)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 校驗(yàn)關(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里會(huì)把整個(gè)文件讀進(jìn)內(nèi)存所以生產(chǎn)環(huán)境必須用hashlib.md5()的增量更新模式見 3.3 節(jié)。這里只是示意邏輯。3.2 如何避免合并過程中的「競態(tài)條件」用文件鎖 臨時(shí)目錄雙保險(xiǎn)多個(gè)用戶同時(shí)上傳同名文件比如都叫report.pdf或同一用戶多次上傳會(huì)導(dǎo)致merge_chunks()被并發(fā)調(diào)用可能產(chǎn)生文件覆蓋。upload-demo 用兩層防護(hù)臨時(shí)目錄隔離每個(gè)文件的分片存放在UPLOAD_FOLDER/chunks/{file_md5}/下file_md5是全局唯一 ID合并鎖文件在merge_chunks()開頭創(chuàng)建.merging鎖文件結(jié)束時(shí)刪除。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)級(jí) MD5 校驗(yàn)用增量哈希避免內(nèi)存溢出上面代碼里f.read()會(huì)把整個(gè)文件加載進(jìn)內(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() # 合并后立即校驗(yàn) final_md5 calculate_file_md5(final_path) if final_md5 ! file_md5: os.remove(final_path) raise Exception(File integrity check failed)這個(gè)iter(lambda: f.read(4096), b)是 Python 標(biāo)準(zhǔn)技巧它會(huì)不斷調(diào)用f.read(4096)直到返回空 bytesb為止完美避開一次性讀取大文件。4. 部署 Flask 分片上傳服務(wù)gunicorn nginx 配置的三個(gè)生死線4.1 nginx 配置必須突破的兩個(gè)默認(rèn)限制Flask 本地測試沒問題一上 nginx 就 413 Request Entity Too Large 或 504 Gateway Timeout。這是因?yàn)?nginx 默認(rèn)限制了client_max_body_size控制單個(gè)請求體最大尺寸分片請求雖小但合并后的大文件上傳可能走另一路proxy_read_timeout控制 nginx 等待 upstreamFlask響應(yīng)的時(shí)間。# /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分片上傳合并可能耗時(shí)較長尤其機(jī)械硬盤 proxy_read_timeout 3600; # 1小時(shí)足夠傳 10GB # 生死線3轉(zhuǎn)發(fā)原始 Host 和 Connection 頭避免 Flask 獲取錯(cuò)誤 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ù)層校驗(yàn)如if file_size 10*1024*1024*1024: abort()否則可能被惡意上傳打爆磁盤。4.2 gunicorn 啟動(dòng)參數(shù)為什么--timeout必須大于proxy_read_timeoutgunicorn 的--timeout是 worker 進(jìn)程處理單個(gè)請求的最大秒數(shù)。如果 nginx 的proxy_read_timeout設(shè)為 3600而 gunicorn 的--timeout是默認(rèn) 30 秒那么上傳進(jìn)行到 31 秒時(shí)gunicorn 就會(huì)殺掉 worker返回 502 Bad Gateway。# 正確啟動(dòng)命令timeout 必須 ≥ nginx proxy_read_timeout gunicorn -w 4 -b 127.0.0.1:8000 --timeout 3600 --keep-alive 5 app:app-w 44 個(gè) worker應(yīng)對并發(fā)分片請求--keep-alive 5保持 HTTP 連接 5 秒減少 TCP 握手開銷分片上傳會(huì)發(fā)幾十個(gè)請求。4.3 Flask 配置文件里必須顯式關(guān)閉的兩個(gè)危險(xiǎn)開關(guān)app.config里有兩個(gè)默認(rèn)開啟的選項(xiàng)在分片上傳場景下必須關(guān)掉# config.py class Config: # 危險(xiǎn)開關(guān)1Werkzeug 的文件緩存會(huì)把小分片暫存內(nèi)存大分片反而觸發(fā)磁盤緩存行為不可控 MAX_CONTENT_LENGTH None # 關(guān)閉全局請求體限制由 nginx 控制 # 危險(xiǎn)開關(guān)2Flask 的 JSON 自動(dòng)解析分片請求是 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。很多翻車案例都是因?yàn)殚_發(fā)者在 Flask 層設(shè)了MAX_CONTENT_LENGTH 100 * 1024 * 1024結(jié)果 nginx 放行了Flask 卻在request.stream.read()時(shí)拋RequestEntityTooLarge。5. 避坑五個(gè)血淚換來的分片上傳排錯(cuò)清單現(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)。簡易方案是加個(gè) 2 秒延時(shí)再提示uploader.on(uploadFinished, function() { setTimeout(() { alert(上傳完成文件正在后臺(tái)合并請稍候...); }, 2000); });5.2 現(xiàn)象同一文件上傳兩次uploads/下出現(xiàn)兩個(gè)不同名字的文件如report.pdf和report (1).pdf原因前端沒傳originalFilename或 Flask 用secure_filename()處理時(shí)中文文件名被截?cái)嗷蜣D(zhuǎn)義導(dǎo)致os.path.join()生成重復(fù)路徑。解決強(qiáng)制前端傳原始文件名并在 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)象合并后文件能打開但播放器報(bào)「moov atom not found」MP4 視頻無法播放原因MP4 文件的元數(shù)據(jù)moov box通常在文件開頭但分片上傳時(shí)被切散合并后 moov 可能落在文件中部。FFmpeg 無法識(shí)別。解決合并后用ffmpeg -i input.mp4 -c copy -movflags faststart output.mp4重寫 moov 到開頭。upload-demo 可加個(gè) 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 報(bào)Failed to load resource: The network connection was lost原因Safari 對XMLHttpRequest的分片上傳有更嚴(yán)格的 CORS 和 header 限制尤其X-Requested-With頭缺失時(shí)。解決在 Flask 中顯式允許所有 headers并關(guān)閉X-Requested-With校驗(yàn)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 配置會(huì) 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. 進(jìn)階技巧用 Redis 做分片狀態(tài)中心實(shí)現(xiàn)跨服務(wù)器合并與實(shí)時(shí)進(jìn)度推送6.1 為什么需要 Redis單機(jī) Flask 部署夠用但集群下分片可能落到不同 worker當(dāng)用gunicorn -w 4或 Kubernetes 多 Pod 時(shí)分片請求可能被負(fù)載均衡到不同 Flask 實(shí)例。A 實(shí)例收到 chunk 0B 實(shí)例收到 chunk 1C 實(shí)例收到 chunk 2……沒有共享存儲(chǔ)合并就會(huì)失敗。upload-demo 原生不支持但加 Redis 只需 20 行代碼。# redis_client.py import redis r redis.Redis(hostlocalhost, port6379, db0) # 存儲(chǔ)分片狀態(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 實(shí)現(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 實(shí)時(shí)進(jìn)度推送用 Server-Sent EventsSSE替代輪詢WebUploader 的進(jìn)度條只顯示當(dāng)前 chunk用戶不知道「合并還要多久」。用 SSE 推送后端真實(shí)進(jìn)度# app.py 新增路由 app.route(/progress/file_md5) def progress_stream(file_md5): def event_stream(): while True: # 從 Redis 讀取當(dāng)前合并進(jìn)度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}%); };從那以后我每次上線新上傳功能都強(qiáng)制走一遍「斷網(wǎng)重傳 → 切換瀏覽器 → 上傳超大文件 → 查看 nginx error.log」四步驗(yàn)證。upload-demo 的價(jià)值不在代碼多精巧而在它把分片上傳里所有能踩的坑都提前用注釋和結(jié)構(gòu)標(biāo)好了。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取