实现分片上传、断点续传、秒传 （JS+NodeJS）（TypeScript） |
您所在的位置：网站首页 › 前端批量上传图片实现顺序上传的方法 › 实现分片上传、断点续传、秒传 （JS+NodeJS）（TypeScript） |
实现分片上传、断点续传、秒传 （JS+NodeJS）（TypeScript）
|
一、引入及效果
上传文件是一个很常见的操作，但是当文件很大时，上传花费的时间会非常长，上传的操作就会具有不确定性，如果不小心连接断开，那么文件就需要重新上传，导致浪费时间和网络资源。 所以，当涉及到大文件上传时，分片上传和断点续传就显得很有必要。把文件分成多个分片，一片片上传，最终服务端再进行合并操作。如果遇到网络中断的问题，已上传的分片就无需上传了，实现断点续传。 使用效果： photo.lhbbb.top/video/slice.mp4 (将分片大小设置为了1*1024B，便于观察) 二、思路 1.文件唯一标识因为需要进行多个分片的上传、续传等操作，我们需要知道这个文件的唯一标识，这样后端才能知道这些分片属于哪个文件，断点续传的时候也能判断这个文件是否曾经上传过。 方法一：可以通过计算文件的md5，来得到文件的唯一标识。（文件被修改过 或 不同文件的唯一标识都是不同的） 优点：文件标识的唯一性可以保证，且有现成的计算md5的库可以调用 缺点：计算文件的md5是一个非常耗时的操作，文件越大，花费的时间越多。JS又是单线程的，计算md5的过程中无法执行其它操作。 方法二：大文件计算md5花费的时间很长，就退而求其次， 使用 文件名+文件最后修改时间+文件大小 作为唯一标识。 优点：计算速度快，因为这些数据在文件的File对象上都有。 缺点：有极小概率可能导致标识重复。 综合方法：可以这样：文件大小低于某个阈值，使用方法一，文件过大时，使用方法二 2.前端1. 计算唯一标识：通过input标签，拿到File文件之后，我们就可以计算唯一标识了。 2. 秒传：计算好唯一标识后，拿着这个唯一标识去请求后端接口，看看该文件是否曾经上传过，如果上传过，就直接显示“上传成功”，也就是实现秒传。 3. 断点续传：如果曾经没有完整上传成功过，就开始对文件进行分片。分好片后，拿着分片的信息和唯一标识，请求后端接口，看看这个文件及其分片是否 “上传了，但没完全上传”，如果符合，就进行断点续传。 （这里我实现的思路是，传给后端分片的总长度，后端去看后端文件夹中有几个分片，返回缺少的分片序号。也可以通过对每个分片进行计算唯一标识，更加稳妥，但是比较费时） 4. 分片上传：如果这个文件完全没有上传过，就开始走正常的分片上传流程。遍历所有分片，把分片通过请求的方式发送到后端。 （注意：浏览器的请求并发数量一般是6个，如果一次性把所有分片请求都发送了，会导致后续的请求需要等待。对此，我们可以使用Promise并发池进行控制，减缓HTTP压力） 5. 完整性校验：全部分片上传完毕后，需要进行完整性校验，避免某个分片遇到故障没有成功上传。发送请求给后端，后端来判断分片数量是否正常，若不正常就上传缺少的分片。 6. 发送合并请求：完整性校验通过后，发送合并请求，后端进行文件合并，返回文件访问路径。 总结： 计算唯一标识 --> 若上传过，则秒传 --> 分片 --> 若上传过部分，则断点续传 --> 正常分片上传 --> 完整性校验 --> 合并请求 3.后端1. 判断文件是否曾经完整上传过：根据文件md5，查看对应目录下的文件是否存在 2. 判断文件是否上传过切片：根据文件md5，查看对应目录中缺少哪些切片 3. 分片上传的接收：把分配存储到一个临时文件夹中，文件夹名字用 唯一标识 命名，方便以后查找。每个分片的命名用分片的序号，便于校验分片是否缺失。 4. 完整性校验：根据md5和分片总数，去检查对应的临时文件夹下缺少哪些文件序号，返回缺少的序号数组。 5. 合并文件：按照分片序号顺序，把分片写入一个新文件中，然后返回这个新文件的访问路径 三、前端实现秒传和断点续传放最后讲，就是一个函数执行顺序的问题。 一些用到的ts类型如下： /**进度条函数的类型，参数是进度，是一个小数，需要 *100 才是正常进度条 */ type onProgress = (progress: number) => any /**切片的类型 */ interface sliceType { /**分片的序号 */ flag: number; /**分片的二进制数据 */ blob: Blob; } 1. 计算文件唯一标识这里计算md5使用了一个库，需要npm安装： npm i spark-md5 /**获得文件的md5，用来作为唯一索引 - 文件过大时，获取md5会很长很长时间，可以考虑用 文件名+文件最后修改时间+文件大小 做“唯一”标识*/ const getMd5 = (file: File) => { return new Promise(async (resolve, reject) => { try { const reader = new FileReader(); reader.readAsArrayBuffer(file); // 当文件读取完成时，计算文件MD5值 reader.onloadend = function (e) { if (!e.target?.result) { return reject('文件读取失败') } const spark = new SparkMD5.ArrayBuffer() spark.append(e.target.result as ArrayBuffer) const md5 = spark.end() resolve(md5) } } catch (error) { reject(error) } }) } 2. 对文件分片File对象的原型对象（Blob）上有一个slice方法，我们可以通过这个来进行分片
传入文件对象和需要的单个分片的大小，单位字节 （比如传入 5*1024*1024 就是5MB） 得到一个分片数组，每个分片包含分片序号和对应的Blob /**文件切片 */ const fileSlice = (file: File, chunkSize: number) => { const result: sliceType[] = [] let index = 0 for (let nowSize = 0; nowSize < file.size; nowSize += chunkSize) { result.push({ flag: index, blob: file.slice(nowSize, nowSize + chunkSize), }) index++ } return result } 3. 把分片上传 /**生成上传切片的promise函数 */ const getSliceUploadPromise = (slice: sliceType, md5: string) => { const formData = new FormData(); formData.append(`file`, slice.blob); formData.append(`index`, String(slice.flag)); formData.append(`md5`, md5); return () => request.postByForm('/sliceUpload/upload', formData) //这里填写自己封装的请求函数 } /**把所有切片上传 */ const sliceUpload = async (sliceList: sliceType[], md5: string, onProgress: onProgress) => { const taskList: (() => Promise)[] = [] const length = sliceList.length for (let i = 0; i < length; i++) { taskList.push(getSliceUploadPromise(sliceList[i], md5)) } //使用并发池优化，避免堵塞 const res = await promisePool(taskList, 5, (count) => onProgress(count / length)) return res }为了避免请求拥塞，这里使用promise并发池进行优化，最大并发数量5。（而不是一次性把所有请求都发出去） /**Promise并发池，当有大量promise并发时，可以通过这个来限制并发数量 * @param taskList 任务列表 * @param max 最大并发数量 * @param oneFinishCallback 每个完成的回调，参数是当前完成的个数和执行结果，可以用来制作进度条 * @retrun 返回每个promise的结果，顺序和任务列表相同。 目前是成功和失败都会放入该结果 * @template T 泛型T会自动填写，是promise成功的结果 * @template Err 此泛型是promise错误的结果 （因为 成功和失败都会放入res，所以加个泛型可以便于ts判断） */ const promisePool = async (taskList: (() => Promise)[], max: number, oneFinishCallback?: (count: number, res: T | Err) => any) => { return new Promise(async (resolve, reject) => { type resType = T | Err try { const length = taskList.length const pool: Promise[] = []//并发池 let count = 0//当前结束了几个 const res = new Array(length) for (let i = 0; i < length; i++) { let task = taskList[i](); //成功和失败都要执行的函数 const handler = (_res: resType) => { pool.splice(pool.indexOf(task), 1) //每当并发池跑完一个任务,从并发池删除个任务 res[i] = _res //放入结果数组 count++ oneFinishCallback && oneFinishCallback(count, _res) if (count === length) { return resolve(res) } } task.then((data) => { handler(data) console.log(`第${i}个任务完成，结果为`, data); }, (err) => { handler(err) console.log(`第${i}个任务失败，原因为`, err); }) pool.push(task); if (pool.length === max) { //Promise.race：返回最快执行的promise //利用Promise.race来看获得哪个任务完成的信号 //搭配await，一旦发现有任务完成了，就继续for循环，把并发池塞满 await Promise.race(pool) } } } catch (error) { console.error('promise并发池出错', error); reject(error) } }) } 4. 完整性校验 （同时适配断点续传）分片上传完毕后，就要进行完整性校验，判断分片是否完整。 getArr函数：发送请求给后端，后端返回缺少的分片序号数组 （返回空数组代表分片完整）完整性校验：根据缺少的分片序号，把缺少的分片继续上传 （这里兼容了断点续传）。同时限制了重试次数，当超过5次仍然没有完整上传，就判断此次上传失败。 /**请求后端，获得缺少的分片序号数组*/ const getArr = async () => (await request.post('/sliceUpload/integrityCheck', { count: sliceList.length, md5 })).missingArr /**完整性校验，缺少的继续上传 （断点续传） */ const integrityCheck = async (sliceList: sliceType[], md5: string, onProgress: onProgress) => { let maxTest = 5 //最大尝试次数，避免无限尝试 /**缺少的序号数组 */ let missingArr: number[] = await getArr() /**总分片数量 */ const sliceListLength = sliceList.length onProgress((sliceListLength - missingArr.length) / sliceListLength)//更新进度条 while (missingArr.length) { const tasks: (() => Promise)[] = [] for (let i = 0; i < missingArr.length; i++) { tasks.push(getSliceUploadPromise(sliceList[missingArr[i]], md5)) } //使用并发池优化请求 await promisePool(tasks, 5, (count) => onProgress((sliceListLength - (missingArr.length - count)) / sliceListLength))//这里的count是缺少的部分中，完成的数量，作为进度条。 missingArr = await getArr() //上传完成后，再次进行完整性校验。 maxTest-- if (maxTest === 0) { return Promise.reject('尝试五次仍未上传成功') } } } 5. 发送合并请求通过完整性校验后，就来到了最后一步，合并切片 （这里主要是后端干活，前端不做过多解释） /**合并切片，拿到路径 */ const merge = async (file: File, md5: string) => { const suffix = getSuffix(file.name) //拿到后缀 const path = await request.post('/sliceUpload/merge', { md5, suffix }) return path } 6.秒传对秒传的判断，在第一步和第二步之间 （计算好唯一标识后就能进行秒传判断了）（这里主要是后端干活，前端不做过多解释） /**秒传 - 判断文件是否上传过，如果上传过了就直接返回文件路径，不用分片即上传 */ const isUploaded = async (file: File, md5: string) => { const res: isUploadedRes = await request.post('/sliceUpload/isUploaded', { md5, suffix }) return res } // 注：关于 /sliceUpload/isUploaded 接口的返回值： type isUploadedRes = { /**是否上传完整了 */ flag: boolean /**如果上传过，路径是什么 （没上传的话为空） */ path: string /**（仅在flag为false有效）是否上传过，但没上传完整？ 为true就走断点续传 （请调用完整性校验接口） */ noComplete: boolean } 7.完整流程搭配上面的函数观看。整体流程需要根据后端给的接口返回值来进行修改，但整体思路不变 /**分片上传 - 只支持单文件 * @param file 文件 * @param chunkSize 一个分片的大小 * @param setTip 可以用于文字提示 * @param onProgress 上传进度的回调，参数是进度 * @returns 上传文件的url */ export async function uploadBySlice(file: File, chunkSize: number, setTip: (tip: string) => any , onProgress: onProgress) { setTip("正在计算md5"); const md5 = await getMd5(file) const isUploadedFlag = await isUploaded(file, md5) if (isUploadedFlag.flag) {//已经上传过，就直接返回路径，实现秒传 onProgress(1) setTip('文件上传成功') return isUploadedFlag.path } else { setTip('正在进行切片') const sliceList = fileSlice(file, chunkSize) console.log('切片', sliceList); if (!isUploadedFlag.noComplete) { //没传递过的文件，才走完整的上传流程 （断点续传） setTip('正在进行文件上传') await sliceUpload(sliceList, md5, onProgress) } else { setTip('正在进行断点续传') } await integrityCheck(sliceList, md5, onProgress)//不管是断点续传还是正常上传，都走这个函数，进行复用 （正常上传的也要校验完整性，断点续传的也要根据校验的完整性来继续上传） setTip("正在合并文件") const path = await merge(file, md5) setTip("文件上传成功！") return path } } 四、后端实现这里使用 NextJS （NodeJS） 做后端，由于各个框架使用不同，仅写出关键步骤 下面代码中使用的一些，在nodejs中关于文件操作的函数： //本文件进行一些I/O操作 import fs from 'fs' import path from 'path' import 'server-only'//代表仅服务端可使用 /**写入Buffer文件在指定路径下。路径不存在时将会创建路径 */ export const writeFile = (filePath: string, buffer: Buffer) => { return new Promise(async (resolve, reject) => { try { const directory = path.dirname(filePath); fs.mkdir(directory, { recursive: true }, (err) => { if (err) { reject(err); } else { fs.writeFile(filePath, buffer, (err) => { if (err) { reject(err); } else { resolve(filePath); } }); } }); } catch (error) { reject(error) } }) } /**删除指定路径的文件 */ export const deleteFile = (path: string) => { return new Promise(async (resolve, reject) => { try { fs.unlink(path, (err) => { if (err) reject(err) else resolve() }) } catch (error) { reject(error) } }) } /**删除指定文件夹及其所有文件。 */ export const deleteFolderRecursive = (folderPath: string) => { if (fs.existsSync(folderPath)) { fs.readdirSync(folderPath).forEach((file) => { const currentPath = `${folderPath}/${file}`; if (fs.lstatSync(currentPath).isDirectory()) { // 递归删除子文件夹 deleteFolderRecursive(currentPath); } else { // 删除文件 fs.unlinkSync(currentPath); } }); // 删除空文件夹 fs.rmdirSync(folderPath); } }; /**在文件末尾追加，不存在的话会新增目录 */ export const appendToFile = (text: string | Buffer, filePath: string, errFn?: (err: NodeJS.ErrnoException | null) => void) => { return new Promise(async (resolve, reject) => { try { const directory = path.dirname(filePath); fs.mkdir(directory, { recursive: true }, (err) => { if (err) { reject(err); return; } fs.appendFile(filePath, text, (err) => { if (err) { errFn?.(err); reject(err); } else { resolve(); } }); }); } catch (error) { reject(error) } }) } /**获得路径文件夹下的所有文件 */ export const getDir = (directoryPath: string) => { return new Promise(async (resolve, reject) => { try { fs.readdir(directoryPath, { withFileTypes: true }, (err, files) => { if (err) { reject(err); return; } resolve(files) }) } catch (error) { reject(error) } }) } /**判断文件(文件夹)是否存在。 */ export const fileExists = (filePath: string): boolean => { return fs.existsSync(filePath); }; 1.判断文件是否曾经上传过/sliceUpload/isUploaded 接口 const { md5, suffix } = await xxxxx() //获取body或query的参数，进行参数校验 const realPath = `https://xxx.com/xxxxxx/${md5}${suffix}` //外界访问的路径 const targetFilePath = `/aaaa/${md5}${suffix}` ; //这里是文件存放的路径 （如果曾经完整上传过） /**是否完整的上传过 */ const flag = fileExists(targetFilePath) /**临时文件夹下是否有这个md5的临时文件夹，有就代表可以进行断点续传 */ const noComplete = fileExists(getAbsPath(`/temp/${md5}`)) return resFn.success({ flag: flag, path: flag ? realPath : "", noComplete }); 2.接收分片上传的文件/sliceUpload/upload 接口 const [files, otherData] = await getFormData(request) //从formdata中取出文件和其它数据 （自行根据框架封装） if (!files[0]) throw '文件不存在' await writeFile(`/temp/${otherData.md5}/${otherData.index}`, files[0]) //存放在 /temp/{md5}/ 目录下 return resFn.success('操作成功'); 3.完整性校验接口/sliceUpload/integrityCheck const { count, md5 } = await xxxx(request) //获取前端传递来的数据 const files = await getDir(`/temp/${md5}/`) //拿到这个临时文件夹下的所有文件 const judgeSet = new Set(Array.from({ length: count }, (k, i) => i)) //根据分片总数，生成一个set （假设分片总数为10，那么这里就是 0,1,2...,10 的一个set） //把文件夹中有的文件序号，从set中删除 files.forEach((file) => { judgeSet.delete(parseInt(file.name)) }) //返回缺少的序号，空数组代表通过完整性校验 return resFn.success({ missingArr: [...judgeSet] }); 4.合并文件接口/sliceUpload/merge const { md5, suffix } = await xxx() //拿到前端传来的参数 const tempDirName = `/temp/${md5}/` //临时文件夹的路径 const files = await getDir(tempDirName)//拿到临时文件夹下的全部文件 files.sort((a, b) => parseInt(a.name) - parseInt(b.name));// 按照数字顺序排序文件名数组 //按照切片顺序，把切片的文件写入新文件 for (let i = 0; i < files.length; i++) { const file = files[i]; const content = fs.readFileSync(`/temp/${md5}/${file.name}`);//切片的位置 await appendToFile(content, `/xxxx/${md5}${suffix}`) // 写入在服务器上存放文件的路径 } deleteFolderRecursive(tempDirName)//删除temp文件夹下的切片文件 const realPath = `http://xxxx.com/xxxxx/${md5}${suffix}`//外界访问的路径 return resFn.success(realPath); 五、总结主要是需要有思路，思路有了就很好写了。上面的思路是自己想的，所以可能有些地方不完善，如果有不对的地方欢迎指出 |
【本文地址】
公司简介
联系我们
今日新闻 |
点击排行 |
|
推荐新闻 |
图片新闻 |
|
专题文章 |