当前位置： 首页 > article >正文

实现分片上传、断点续传、秒传 （JS+NodeJS）（TypeScript）

article 2024/12/28 20:46:39

一、引入及效果

        上传文件是一个很常见的操作，但是当文件很大时，上传花费的时间会非常长，上传的操作就会具有不确定性，如果不小心连接断开，那么文件就需要重新上传，导致浪费时间和网络资源。

        所以，当涉及到大文件上传时，分片上传和断点续传就显得很有必要。把文件分成多个分片，一片片上传，最终服务端再进行合并操作。如果遇到网络中断的问题，已上传的分片就无需上传了，实现断点续传。

        使用效果：   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<string>(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<string>)[] = []
  const length = sliceList.length
  for (let i = 0; i < length; i++) {
    taskList.push(getSliceUploadPromise(sliceList[i], md5))
  }
  //使用并发池优化，避免堵塞
  const res = await promisePool<string, string>(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 <T, Err>(taskList: (() => Promise<T>)[], max: number, oneFinishCallback?: (count: number, res: T | Err) => any) => {
    return new Promise<Array<T | Err>>(async (resolve, reject) => {
        type resType = T | Err
        try {
            const length = taskList.length
            const pool: Promise<resType>[] = []//并发池 
            let count = 0//当前结束了几个
            const res = new Array<resType>(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<string>)[] = []
    for (let i = 0; i < missingArr.length; i++) {
      tasks.push(getSliceUploadPromise(sliceList[missingArr[i]], md5))
    }
    //使用并发池优化请求
    await promisePool<string, string>(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<string>(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<void>(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<void>(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<fs.Dirent[]>(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<isUploadedRes>({
    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);

五、总结

        主要是需要有思路，思路有了就很好写了。上面的思路是自己想的，所以可能有些地方不完善，如果有不对的地方欢迎指出


http://www.kler.cn/a/108333.html

相关文章：

  • 在国产电脑上运行PDFSAM软件使用pdf分割合并交替混合处理pdf文档
  • 【HarmonyOS】鸿蒙将资源文件夹Resource-RawFile下的文件存放到沙箱目录下
  • 游戏引擎学习第62天
  • MATLAB符号计算-符号表达式基础运算操作
  • LLM漫谈（八）| OpenAI 12天直播集锦
  • springboot使用自定义的线程池 完成 多线程执行网络请求，返回数据后，统一返回给前段
  • 视频上的水印文字如何去掉?
  • QT webengine显示HTML简单示例
  • was下log4j设置日志不输出问题
  • 【学习笔记】记录一个win 11 操作文件卡顿，Windows 资源管理器CPU占用飙升问题
  • Python对json文件的读取和处理
  • Netty对接阿里云语音识别和录音识别
  • DevOps与CI/CD的最佳实践
  • python随手小练10（南农作业题）
  • 【赠书活动】从瀑布模式到水母模式：ChatGPT如何赋能软件研发全流程
  • 79 电话号码的字母组合
  • 【Linux】解决缓存锁问题：无法获得锁 /var/lib/dpkg/lock-frontend
  • Qt配置OpenCV教程，亲测已试过
  • 如何使用 nvm-windows 这个工具来管理你电脑上的Node.js版本
  • uniapp vue国际化 i18n
  • springcloud技术栈以及相关组件
  • 计算机中了locked勒索病毒怎么办，locked勒索病毒解密，数据恢复
  • SpringSecurity 认证实战
  • 拉扎维模拟CMOS集成电路设计西交张鸿老师课程P2~5视频学习记录
  • c++ 线程安全的string类
  • 雨云虚拟主机使用教程WordPress博客网站搭建教程