分片上傳并不是什么新概念，尤其是大文件傳輸?shù)奶幚碇薪?jīng)常會(huì)被使用，在之前的一篇文章里：python花式讀取大文件(10g/50g/1t)遇到的性能問(wèn)題（面試向）我們討論了如何讀寫(xiě)超大型文件，本次再來(lái)探討一下如何上傳超大型文件，其實(shí)原理都是大同小異，原則就是化整為零，將大文件進(jìn)行分片處理，切割成若干小文件，隨后為每個(gè)分片創(chuàng)建一個(gè)新的臨時(shí)文件來(lái)保存其內(nèi)容，待全部分片上傳完畢后，后端再按順序讀取所有臨時(shí)文件的內(nèi)容，將數(shù)據(jù)寫(xiě)入新文件中，最后將臨時(shí)文件再刪掉。大體流程請(qǐng)見(jiàn)下圖：

其實(shí)現(xiàn)在市面上有很多前端的三方庫(kù)都集成了分片上傳的功能，比如百度的WebUploader，遺憾的是它已經(jīng)淡出歷史舞臺(tái)，無(wú)人維護(hù)了。現(xiàn)在比較推薦主流的庫(kù)是vue-simple-uploader，不過(guò)餓了么公司開(kāi)源的elementUI市場(chǎng)占有率還是非常高的，但其實(shí)大家所不知道的是，這個(gè)非常著名的前端UI庫(kù)也已經(jīng)許久沒(méi)人維護(hù)了，Vue3.0版本出來(lái)這么久了，也沒(méi)有做適配，由此可見(jiàn)大公司的開(kāi)源產(chǎn)品還是需要給業(yè)務(wù)讓步。本次我們利用elementUI的自定義上傳結(jié)合后端的網(wǎng)紅框架FastAPI來(lái)實(shí)現(xiàn)分片上傳。

首先前端需要安裝需要的庫(kù)：

npm install element-ui --save npm install spark-md5 --save npm install axios --save

隨后在入口文件main.js中進(jìn)行配置：

import ElementUI from 'element-ui' import 'element-ui/lib/theme-chalk/index.css' Vue.use(ElementUI) import Axios from 'axios' Vue.prototype.axios = Axios; import QS from 'qs' Vue.prototype.qs = QS;

配置好之后，設(shè)計(jì)方案，前端通過(guò)elementUI上傳時(shí)，通過(guò)分片大小的閾值對(duì)文件進(jìn)行切割，并且記錄每一片文件的切割順序(chunk)，在這個(gè)過(guò)程中，通過(guò)SparkMD5來(lái)計(jì)算文件的唯一標(biāo)識(shí)(防止多個(gè)文件同時(shí)上傳的覆蓋問(wèn)題identifier)，在每一次分片文件的上傳中，會(huì)將分片文件實(shí)體，切割順序（chunk）以及唯一標(biāo)識(shí)（identifier）異步發(fā)送到后端接口(fastapi)，后端將chunk和identifier結(jié)合在一起作為臨時(shí)文件寫(xiě)入服務(wù)器磁盤(pán)中，當(dāng)前端將所有的分片文件都發(fā)送完畢后，最后請(qǐng)求一次后端另外一個(gè)接口，后端將所有文件合并。

根據(jù)方案，前端建立chunkupload.js文件：

import SparkMD5 from 'spark-md5'//錯(cuò)誤信息 function getError(action, option, xhr) { let msg if (xhr.response) { msg = `${xhr.response.error || xhr.response}` } else if (xhr.responseText) { msg = `${xhr.responseText}` } else { msg = `fail to post ${action} ${xhr.status}` } const err = new Error(msg) err.status = xhr.status err.method = 'post' err.url = action return err } // 上傳成功完成合并之后，獲取服務(wù)器返回的json function getBody(xhr) { const text = xhr.responseText || xhr.response if (!text) { return text } try { return JSON.parse(text) } catch (e) { return text } } // 分片上傳的自定義請(qǐng)求，以下請(qǐng)求會(huì)覆蓋element的默認(rèn)上傳行為 export default function upload(option) { if (typeof XMLHttpRequest === 'undefined') { return } const spark = new SparkMD5.ArrayBuffer()// md5的ArrayBuffer加密類(lèi) const fileReader = new FileReader()// 文件讀取類(lèi) const action = option.action // 文件上傳上傳路徑 const chunkSize = 1024 * 1024 * 1 // 單個(gè)分片大小，這里測(cè)試用1m let md5 = ''// 文件的唯一標(biāo)識(shí) const optionFile = option.file // 需要分片的文件 let fileChunkedList = [] // 文件分片完成之后的數(shù)組 const percentage = [] // 文件上傳進(jìn)度的數(shù)組，單項(xiàng)就是一個(gè)分片的進(jìn)度 // 文件開(kāi)始分片，push到fileChunkedList數(shù)組中， 并用第一個(gè)分片去計(jì)算文件的md5 for (let i = 0; i < optionFile.size; i = i + chunkSize) { const tmp = optionFile.slice(i, Math.min((i + chunkSize), optionFile.size)) if (i === 0) { fileReader.readAsArrayBuffer(tmp) } fileChunkedList.push(tmp) } // 在文件讀取完畢之后，開(kāi)始計(jì)算文件md5，作為文件唯一標(biāo)識(shí) fileReader.onload = async (e) => { spark.append(e.target.result) md5 = spark.end() + new Date().getTime() console.log('文件唯一標(biāo)識(shí)--------', md5) // 將fileChunkedList轉(zhuǎn)成FormData對(duì)象，并加入上傳時(shí)需要的數(shù)據(jù) fileChunkedList = fileChunkedList.map((item, index) => { const formData = new FormData() if (option.data) { // 額外加入外面?zhèn)魅氲膁ata數(shù)據(jù) Object.keys(option.data).forEach(key => { formData.append(key, option.data[key]) }) // 這些字段看后端需要哪些，就傳哪些，也可以自己追加額外參數(shù) formData.append(option.filename, item, option.file.name)// 文件 formData.append('chunkNumber', index + 1)// 當(dāng)前文件塊 formData.append('chunkSize', chunkSize)// 單個(gè)分塊大小 formData.append('currentChunkSize', item.size)// 當(dāng)前分塊大小 formData.append('totalSize', optionFile.size)// 文件總大小 formData.append('identifier', md5)// 文件標(biāo)識(shí) formData.append('filename', option.file.name)// 文件名 formData.append('totalChunks', fileChunkedList.length)// 總塊數(shù) } return { formData: formData, index: index } }) // 更新上傳進(jìn)度條百分比的方法 const updataPercentage = (e) => { let loaded = 0// 當(dāng)前已經(jīng)上傳文件的總大小 percentage.forEach(item => { loaded += item }) e.percent = loaded / optionFile.size * 100 option.onProgress(e) } // 創(chuàng)建隊(duì)列上傳任務(wù)，limit是上傳并發(fā)數(shù)，默認(rèn)會(huì)用兩個(gè)并發(fā) function sendRequest(chunks, limit = 2) { return new Promise((resolve, reject) => { const len = chunks.length let counter = 0 let isStop = false const start = async () => { if (isStop) { return } const item = chunks.shift() console.log() if (item) { const xhr = new XMLHttpRequest() const index = item.index // 分片上傳失敗回調(diào) xhr.onerror = function error(e) { isStop = true reject(e) } // 分片上傳成功回調(diào) xhr.onload = function onload() { if (xhr.status < 200 || xhr.status >= 300) { isStop = true reject(getError(action, option, xhr)) } if (counter === len - 1) { // 最后一個(gè)上傳完成 resolve() } else { counter++ start() } } // 分片上傳中回調(diào) if (xhr.upload) { xhr.upload.onprogress = function progress(e) { if (e.total > 0) { e.percent = e.loaded / e.total * 100 } percentage[index] = e.loaded console.log(index) updataPercentage(e) } } xhr.open('post', action, true) if (option.withCredentials && 'withCredentials' in xhr) { xhr.withCredentials = true } const headers = option.headers || {} for (const item in headers) { if (headers.hasOwnProperty(item) && headers[item] !== null) { xhr.setRequestHeader(item, headers[item]) } } // 文件開(kāi)始上傳 xhr.send(item.formData); } } while (limit > 0) { setTimeout(() => { start() }, Math.random() * 1000) limit -= 1 } }) } try { // 調(diào)用上傳隊(duì)列方法 等待所有文件上傳完成 await sendRequest(fileChunkedList,2) // 這里的參數(shù)根據(jù)自己實(shí)際情況寫(xiě) const data = { identifier: md5, filename: option.file.name, totalSize: optionFile.size } // 給后端發(fā)送文件合并請(qǐng)求 const fileInfo = await this.axios({ method: 'post', url: 'http://localhost:8000/mergefile/', data: this.qs.stringify(data) }, { headers: { "Content-Type": "multipart/form-data" } }).catch(error => { console.log("ERRRR:: ", error.response.data); }); console.log(fileInfo); if (fileInfo.data.code === 200) { const success = getBody(fileInfo.request) option.onSuccess(success) return } } catch (error) { option.onError(error) } } }

之后建立upload.vue模板文件，并且引入自定義上傳控件：

<template> <div> <el-upload :http-request="chunkUpload" :ref="chunkUpload" :action="uploadUrl" :data="uploadData" :on-error="onError" :before-remove="beforeRemove" name="file" > <el-button size="small" type="primary">點(diǎn)擊上傳</el-button> </el-upload> </div> </template> <script> //js部分 import chunkUpload from './chunkUpload' export default { data() { return { uploadData: { //這里面放額外攜帶的參數(shù) }, //文件上傳的路徑 uploadUrl: 'http://localhost:8000/uploadfile/', //文件上傳的路徑 chunkUpload: chunkUpload // 分片上傳自定義方法，在頭部引入了 } }, methods: { onError(err, file, fileList) { this.$store.getters.chunkUploadXhr.forEach(item => { item.abort() }) this.$alert('文件上傳失敗，請(qǐng)重試', '錯(cuò)誤', { confirmButtonText: '確定' }) }, beforeRemove(file) { // 如果正在分片上傳，則取消分片上傳 if (file.percentage !== 100) { this.$store.getters.chunkUploadXhr.forEach(item => { item.abort() }) } } } } </script> <style> </style>

這里定義的后端上傳接口是：http://localhsot:8000/uploadfile/ 合并文件接口是：http://localhsot:8000/mergefile/

此時(shí)啟動(dòng)前端的vue.js服務(wù)：

npm run dev

頁(yè)面效果見(jiàn)下圖：

前端搞定了，下面我們來(lái)編寫(xiě)接口，后端的任務(wù)相對(duì)簡(jiǎn)單，利用FastAPI接收分片文件、分片順序以及唯一標(biāo)識(shí)，并且將文件臨時(shí)寫(xiě)入到服務(wù)器中，當(dāng)最后一個(gè)分片文件完成上傳后，第二個(gè)接口負(fù)責(zé)按照分片順序合并所有文件，合并成功后再刪除臨時(shí)文件，用來(lái)節(jié)約空間，先安裝依賴(lài)的三方庫(kù)

pip3 install python-multipart

當(dāng)然了，由于是前后端分離項(xiàng)目，別忘了設(shè)置一下跨域，編寫(xiě)main.py:

from uploadfile import router from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from model import database from fastapi.middleware.cors import CORSMiddleware app = FastAPI() origins = [ "*" ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.mount("/static", StaticFiles(directory="static"), name="static") templates = Jinja2Templates(directory="templates") app.include_router(router) @app.on_event("startup") async def startup(): await database.connect() @app.on_event("shutdown") async def shutdown(): await database.disconnect() @app.get("/") def read_root(): return {"Hello": "World"}

然后編寫(xiě)uploadfile.py:

@router.post("/uploadfile/") async def uploadfile(file: UploadFile = File(...), chunkNumber: str = Form(...), identifier: str = Form(...)): task = identifier # 獲取文件唯一標(biāo)識(shí)符 chunk = chunkNumber # 獲取該分片在所有分片中的序號(hào) filename = '%s%s' % (task,chunk) # 構(gòu)成該分片唯一標(biāo)識(shí)符 contents = await file.read() #異步讀取文件 with open('./static/upload/%s' % filename, "wb") as f: f.write(contents) print(file.filename) return {"filename": file.filename} @router.post("/mergefile/") async def uploadfile(identifier: str = Form(...), filename: str = Form(...)): target_filename = filename # 獲取上傳文件的文件名 task = identifier # 獲取文件的唯一標(biāo)識(shí)符 chunk = 1 # 分片序號(hào) with open('./static/upload/%s' % target_filename, 'wb') as target_file: # 創(chuàng)建新文件 while True: try: filename = './static/upload/%s%d' % (task,chunk) # 按序打開(kāi)每個(gè)分片 source_file = open(filename, 'rb') # 讀取分片內(nèi)容寫(xiě)入新文件 target_file.write(source_file.read()) source_file.close() except IOError: break chunk += 1 os.remove(filename) return {"code":200}

值得一提的是這里我們使用UploadFile來(lái)定義文件參數(shù)，它的優(yōu)勢(shì)在于在接收存儲(chǔ)文件過(guò)程中如果文件過(guò)大超過(guò)了內(nèi)存限制就會(huì)存儲(chǔ)在硬盤(pán)中，相當(dāng)靈活，同時(shí)配合await關(guān)鍵字異步讀取文件內(nèi)容，提高了性能和效率。

啟動(dòng)后端服務(wù)測(cè)試一下效果：

uvicorn main:app --reload

可以看到，當(dāng)我們上傳一張2.9m的圖片時(shí)，前端會(huì)根據(jù)設(shè)置好的的分片閾值將該圖片切割為四份，傳遞給后端接口uploadfile后，后端在根據(jù)參數(shù)用接口mergefile將其合并，整個(gè)過(guò)程一氣呵成、行云流水、勢(shì)如破竹，讓人用了之后禁不住心曠神怡、把酒臨風(fēng)。最后奉上項(xiàng)目地址：https://gitee.com/QiHanXiBei/fastapi_blog

總結(jié)

以上是生活随笔為你收集整理的聚是一团火散作满天星，前端Vue.js+elementUI结合后端FastAPI实现大文件分片上传的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。