簡(jiǎn)介這是一份基于Flask與WebUploader實(shí)現(xiàn)的大文件分片上傳與流式下載的完整Web開(kāi)發(fā)示例面向Python后端初學(xué)者、Web全棧學(xué)習(xí)者及需要解決大文件傳輸瓶頸的開(kāi)發(fā)者。資源通過(guò)前端分片上傳、服務(wù)端分片接收與合并、多用戶(hù)隔離存儲(chǔ)、內(nèi)存友好的流式響應(yīng)等機(jī)制切實(shí)應(yīng)對(duì)傳統(tǒng)單次HTTP上傳易超時(shí)、OOM、并發(fā)沖突等典型問(wèn)題。壓縮包共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頁(yè)面提供上傳入口與文件列表另有字體、圖標(biāo)及requirements依賴(lài)清單整體僅427KB輕量易部署。已有651人學(xué)習(xí)下載讀者可直接運(yùn)行體驗(yàn)分片上傳成功/失敗狀態(tài)、實(shí)時(shí)進(jìn)度條渲染、服務(wù)端分片合并流程及低內(nèi)存占用的流式下載效果同時(shí)獲得符合BSD/MIT協(xié)議規(guī)范的開(kāi)源集成實(shí)踐參考。1. 分段上傳不是“把大文件切成小塊”就完事upload-demo 是一套能扛住 2GB 視頻、斷點(diǎn)續(xù)傳不丟數(shù)據(jù)、前端拖拽后端流式寫(xiě)入的 Flask 實(shí)戰(zhàn)樣板你試過(guò)用普通input typefile上傳一個(gè) 1.8GB 的監(jiān)控錄像嗎瀏覽器卡死、504 網(wǎng)關(guān)超時(shí)、Flask 默認(rèn) 16MB 請(qǐng)求體限制直接報(bào)錯(cuò)——這不是配置調(diào)得不夠狠而是架構(gòu)錯(cuò)了。upload-demo 就是為這種真實(shí)場(chǎng)景而生它用 WebUploader 做前端分片控制支持拖拽、暫停、續(xù)傳、進(jìn)度條用 Flask 做輕量后端不依賴(lài) Celery、不硬編碼路徑、不寫(xiě)死文件名核心邏輯是「接收分片 → 合并校驗(yàn) → 流式落盤(pán)」。它不是玩具 demo而是我在線(xiàn)上教育平臺(tái)部署時(shí)抄過(guò)的底板——學(xué)生上傳 4K 課程視頻、教師批量導(dǎo)入 300 個(gè) PPT 附件都靠它穩(wěn)住。適合正在用 Flask 開(kāi)發(fā)內(nèi)容管理后臺(tái)、教學(xué)系統(tǒng)、醫(yī)療影像歸檔模塊的工程師不適合純靜態(tài)頁(yè)面或已用 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ì)繞過(guò) HTTP 協(xié)議瓶頸把「?jìng)鬏斂煽啃浴箯?TCP 層移到應(yīng)用層HTTP/1.1 本身不支持?jǐn)帱c(diǎn)續(xù)傳。你用fetch()發(fā)一個(gè) 2GB 的 Blob中間網(wǎng)絡(luò)抖動(dòng)一次整個(gè)請(qǐng)求就失敗重試就得從頭傳。WebUploader 的分片機(jī)制本質(zhì)是把一個(gè)大文件切成固定大小如 5MB的 chunk每個(gè) chunk 單獨(dú)發(fā) POST 請(qǐng)求帶唯一標(biāo)識(shí)chunkIndex、totalChunks、fileMd5。服務(wù)端收到后先校驗(yàn)fileMd5 chunkIndex是否已存在再?zèng)Q定是跳過(guò)、覆蓋還是追加。這相當(dāng)于在應(yīng)用層實(shí)現(xiàn)了「TCP 的 ACK 機(jī)制」丟一個(gè) chunk只重傳那一個(gè)不影響其他。Flask 本身沒(méi)提供分片 API但它的 request.stream 和 werkzeug.utils.secure_filename 讓我們能安全地接收、命名、拼接這些 chunk——這才是 upload-demo 的技術(shù)錨點(diǎn)不是炫技是剛需。2.2 Flask 端為何不用request.files流式接收才是處理大文件的唯一正解很多新手一上來(lái)就寫(xiě)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)前分片全部?jī)?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)制寫(xiě)入不經(jīng)過(guò)內(nèi)存緩沖 with open(chunk_path, wb) as f: f.write(chunk_data) return jsonify({status: success, chunk: chunk_index})提示request.stream.read()會(huì)阻塞直到客戶(hù)端發(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 后端對(duì)齊分片參數(shù)四個(gè)必須同步的字段WebUploader 的server配置項(xiàng)必須和 Flask 路由嚴(yán)格匹配否則分片會(huì) 400。以下是實(shí)際調(diào)試中驗(yàn)證過(guò)的最小可行配置// static/js/upload.js var uploader WebUploader.create({ // 1. 分片大小后端合并時(shí)按此切分建議 2-10MB太小增加 HTTP 開(kāi)銷(xiāo)太大失去斷點(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)必須開(kāi)啟否則無(wú)法判斷重復(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寫(xiě)成chunk_index導(dǎo)致后端收不到索引所有分片都存到00000文件里——合并時(shí)順序全亂。3. 合并分片不是簡(jiǎn)單cat *.bin final.file校驗(yàn)、去重、原子寫(xiě)入三步缺一不可3.1 合并前必須做文件級(jí) MD5 校驗(yàn)為什么不能只信chunkIndex順序分片上傳最隱蔽的坑是「網(wǎng)絡(luò)亂序」客戶(hù)端可能先發(fā) chunk 5再發(fā) chunk 1尤其在弱網(wǎng)下。如果后端只按chunkIndex數(shù)字排序合并得到的文件就是亂碼。upload-demo 的解決方案是合并前先檢查所有分片是否齊全再按chunkIndex排序最后用hashlib.md5()對(duì)合并后的完整文件做二次校驗(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()的增量更新模式見(jiàn) 3.3 節(jié)。這里只是示意邏輯。3.2 如何避免合并過(guò)程中的「競(jìng)態(tài)條件」用文件鎖 臨時(shí)目錄雙保險(xiǎn)多個(gè)用戶(hù)同時(shí)上傳同名文件比如都叫report.pdf或同一用戶(hù)多次上傳會(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()開(kāi)頭創(chuàng)建.merging鎖文件結(jié)束時(shí)刪除。lock_file os.path.join(chunk_dir, .merging) if os.path.exists(lock_file): time.sleep(0.1) # 簡(jiǎn)單退避生產(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為止完美避開(kāi)一次性讀取大文件。4. 部署 Flask 分片上傳服務(wù)gunicorn nginx 配置的三個(gè)生死線(xiàn)4.1 nginx 配置必須突破的兩個(gè)默認(rèn)限制Flask 本地測(cè)試沒(méi)問(wèn)題一上 nginx 就 413 Request Entity Too Large 或 504 Gateway Timeout。這是因?yàn)?nginx 默認(rèn)限制了client_max_body_size控制單個(gè)請(qǐng)求體最大尺寸分片請(qǐng)求雖小但合并后的大文件上傳可能走另一路proxy_read_timeout控制 nginx 等待 upstreamFlask響應(yīng)的時(shí)間。# /etc/nginx/sites-available/upload-demo server { listen 80; server_name upload.example.com; # 生死線(xiàn)1必須設(shè)為 0無(wú)限制或大于你最大單文件尺寸 client_max_body_size 0; location / { proxy_pass http://127.0.0.1:8000; # 生死線(xiàn)2分片上傳合并可能耗時(shí)較長(zhǎng)尤其機(jī)械硬盤(pán) proxy_read_timeout 3600; # 1小時(shí)足夠傳 10GB # 生死線(xiàn)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()否則可能被惡意上傳打爆磁盤(pán)。4.2 gunicorn 啟動(dòng)參數(shù)為什么--timeout必須大于proxy_read_timeoutgunicorn 的--timeout是 worker 進(jìn)程處理單個(gè)請(qǐng)求的最大秒數(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)對(duì)并發(fā)分片請(qǐng)求--keep-alive 5保持 HTTP 連接 5 秒減少 TCP 握手開(kāi)銷(xiāo)分片上傳會(huì)發(fā)幾十個(gè)請(qǐng)求。4.3 Flask 配置文件里必須顯式關(guān)閉的兩個(gè)危險(xiǎn)開(kāi)關(guān)app.config里有兩個(gè)默認(rèn)開(kāi)啟的選項(xiàng)在分片上傳場(chǎng)景下必須關(guān)掉# config.py class Config: # 危險(xiǎn)開(kāi)關(guān)1Werkzeug 的文件緩存會(huì)把小分片暫存內(nèi)存大分片反而觸發(fā)磁盤(pán)緩存行為不可控 MAX_CONTENT_LENGTH None # 關(guān)閉全局請(qǐng)求體限制由 nginx 控制 # 危險(xiǎn)開(kāi)關(guān)2Flask 的 JSON 自動(dòng)解析分片請(qǐng)求是 multipart/form-data不是 JSON JSONIFY_PRETTYPRINT_REGULAR False JSON_SORT_KEYS False # 安全加固上傳目錄絕對(duì)路徑禁止相對(duì)路徑 UPLOAD_FOLDER /var/www/upload-demo/uploadsMAX_CONTENT_LENGTH None是關(guān)鍵——它讓 Flask 不再攔截request.stream把控制權(quán)完全交給 nginx。很多翻車(chē)案例都是因?yàn)殚_(kāi)發(fā)者在 Flask 層設(shè)了MAX_CONTENT_LENGTH 100 * 1024 * 1024結(jié)果 nginx 放行了Flask 卻在request.stream.read()時(shí)拋RequestEntityTooLarge。5. 避坑五個(gè)血淚換來(lái)的分片上傳排錯(cuò)清單現(xiàn)象→原因→解決5.1 現(xiàn)象前端顯示「上傳完成」但服務(wù)器uploads/目錄下沒(méi)有文件原因WebUploader 的onUploadFinished回調(diào)觸發(fā)過(guò)早未等后端合并完成就結(jié)束。upload-demo 的/merge接口是異步的但前端沒(méi)監(jiān)聽(tīng)合并結(jié)果。解決在 WebUploader 的onUploadFinished里加輪詢(xún)或 WebSocket 監(jiān)聽(tīng)合并狀態(tài)。簡(jiǎn)易方案是加個(gè) 2 秒延時(shí)再提示uploader.on(uploadFinished, function() { setTimeout(() { alert(上傳完成文件正在后臺(tái)合并請(qǐng)稍候...); }, 2000); });5.2 現(xiàn)象同一文件上傳兩次uploads/下出現(xiàn)兩個(gè)不同名字的文件如report.pdf和report (1).pdf原因前端沒(méi)傳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)象合并后文件能打開(kāi)但播放器報(bào)「moov atom not found」MP4 視頻無(wú)法播放原因MP4 文件的元數(shù)據(jù)moov box通常在文件開(kāi)頭但分片上傳時(shí)被切散合并后 moov 可能落在文件中部。FFmpeg 無(wú)法識(shí)別。解決合并后用ffmpeg -i input.mp4 -c copy -movflags faststart output.mp4重寫(xiě) moov 到開(kāi)頭。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 對(duì)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 里有下劃線(xiàn)WebUploader 用 location /upload { proxy_pass http://127.0.0.1:8000/upload; proxy_pass_request_headers on; # 必須開(kāi)啟 }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í)分片請(qǐng)求可能被負(fù)載均衡到不同 Flask 實(shí)例。A 實(shí)例收到 chunk 0B 實(shí)例收到 chunk 1C 實(shí)例收到 chunk 2……沒(méi)有共享存儲(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)「收到最后一片立刻合并」不再輪詢(xún)數(shù)據(jù)庫(kù)而是用 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替代輪詢(xún)WebUploader 的進(jìn)度條只顯示當(dāng)前 chunk用戶(hù)不知道「合并還要多久」。用 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)聽(tīng)const eventSource new EventSource(/progress/${fileMd5}); eventSource.onmessage function(e) { const progress parseInt(e.data); $(#merge-progress).text(合并中${progress}%); };從那以后我每次上線(xiàn)新上傳功能都強(qiáng)制走一遍「斷網(wǎng)重傳 → 切換瀏覽器 → 上傳超大文件 → 查看 nginx error.log」四步驗(yàn)證。upload-demo 的價(jià)值不在代碼多精巧而在它把分片上傳里所有能踩的坑都提前用注釋和結(jié)構(gòu)標(biāo)好了。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取