原生 JavaScript 实现大文件分片并发上传

时间：2021-2-20 作者：admin

原生 JavaScript 实现大文件分片并发上传

首先放上github链接，代码注释的比较清除，上层使用React进行测试，基本的功能都实现了

技术栈 TypeScript node

模块 sprak-md5 express fs fromidadle

实现的功能

  • 分片上传
  • 验证秒传
  • 断点续传
  • 并发上传

前言

此项目本来是作为一个react上传组件编写，编写后发现该组件内部相对比较复杂，于是将该组件抽离出来写成了一个上传工具类

由于分片上传需要后端支持，所以本片会涉及到一点node的知识

为什么需要文件分片上传？

在上传小文件时，分片上传和普通上传的效果体验并不大，但是在上传大文件时，普通的一次性上传会存在以下缺点

  • 文件上传时间较长，会长时间持续占用服务器的连接端口
  • 如果断网或者页面不小心关闭，已上传的文件会全部丢失，需要重新上传

分片上传的优点

  • 充分利用浏览器多进程的特性，并发的上传文件，加快文件的上传速度
  • 服务端可以将已经上传的文件切片保存起来，若文件上传过程中出现了意外，再下次上传的时候可以过滤掉已经上传的切片

核心思想

利用H5提供的原生File对象，由于File对象是特殊类型的Blob, File接口也继承了Blob接口的属性，分片上传的核心思想就是利用File继承Blob接口的Blob.slice方法

Blob.slice方法可以将我们的文件切分为多个单个的切片，分片上传的思想就是利用slice APi将文件分割成多个切片，然后利用浏览器多进程的特性进行并发上传

如何实现验证秒传？

思路

在上传切片前将文件的基本信息发送至服务端进行验证，判断该文件是否需要重新上传

验证方法

  • 文件名
  • 文件最后修改的时间（File文件对象的一个属性 lastModified）
  • 文件hash值

如何选择

每一验证的方法都有其优缺点，文件名和最后修改的时间方法相较于计算文件hash值可以在前端逻辑中快速的判断该文件是否需要上传，但是文件名和最后修改的时间无法准确的判断该文件是否需要重传，因为文件名的修改对文件内容无影响，而lastModified又过于“敏感”，即对文件内容无实质修改的操作也会被记录，从而导致该已上传的文件需要重新上传。

所以我们选用计算文件hash值的方法判断文件是否需要上传，每次上传文件前首先将文件的hash值发送至服务端判断该文件是否需要重新上传，hash的计算我们选用spark-md5，别问为啥，因为spark-md5已经算是比较快的一个计算hash值的库了

spark-md5

spark-md5是一个计算文件hash值的工具

spark-md5可以帮助我们相对比较快速的计算出文件的hash值，我们可以在服务端存储文件的hash的

spark-md5基本用法

// 传入的参数为文件切片数组
// 首先递归的将每一个切片可利用FileReader实例读取文件中的内容，在成功的回调函数中将文件添加到spark-md实例中，在递归结束时计算文件hash，并返回文件hash，考虑到该过程比较耗，所以使用promise进行包裹

const calculatehash = (fileChunkList: Array<chunkListsFile>) => (
    new Promise(reslove => {
        const spark = new SparkMD5.ArrayBuffer()
        let count = 0
        const loadNext = (index: number) => {
            const reader = new FileReader()
            reader.readAsArrayBuffer(fileChunkList[index].file)
            reader.onload = (e: any) => {
                count++
                spark.append(e.target.result)
                // 如果文件处理完成则发送发送请求
                if (count === fileChunkList.length) {
                    reslove(spark.end())
                    return
                }
                loadNext(count)
            }
        }
        loadNext(0)
    })
)

实现

前端

因为我们是作为一个组件去写的，所以我们默认不处理html获取文件对象的步骤，直接从传入文件对象列表开始写，在文件的处理过程中调用传入的参数上报当前的文件处理/上传进度

接口定义

由于开发过程是由ts进行开发，所以我们首先要先定义接口

接口名 描述
fileBasicMessage 文件基本类型
chunkListsFile 每个文件切片
IwaitUploadFile 待上传文件数组
IwaitCalculateFile 待计算hash文件数组
IuploadedFile 上传完成的文件数组 
interface fileBasicMessage {
    file: File,
    id?: string
}


export interface chunkListsFile {
    file: Blob
    hash: string
    fileName: string
    index?: number
}

export interface IwaitUploadFile extends fileBasicMessage {
    hash?: string
    uploadProcess?: number
    uploadPercentArr: Array<number> | []
    chunkList: Array<chunkListsFile> | []
    uploadedSize: number
}

export interface IwaitCalculateFile {
    id: string
    file: File
}

export interface IuploadedFile {
    url: string
    fileName: string
}

UML图

工具类大体框架实现

构造一个文件上传类并编写添加文件方法

//  函数需要的参数
export interface Iprops {
    // 每个切片的大小
    chunkSize?: number
    //  每个文件切片列表允许并发上传的个数
    concurrency: number
    //  上报文件处理上传进度回调函数
    updateWaitCalculateFile: (files: Array<IwaitCalculateFile>) => void
    updateWaitUploadFile: (files: Array<IwaitUploadFile>) => void
    updateUploadedFiles: (files: Array<IuploadedFile>) => void
}

class UploadTool {

    // 首先定义我们需要的信息
    constructor(props: IProps) {
        // 是否正在计算文件hash
        this.isCalculating = false
        // 切片大小默认4M
        this.chunkSize = props.chunkSize ? props.chunkSize : 4 * 1024 * 1024
        this.concurrency = props.concurrency ? props.concurrency : 3
        this.updateWaitCalculateFile = props.updateWaitCalculateFile
        this.updateWaitUploadFile = props.updateWaitUploadFile
        this.updateUploadedFiles = props.updateUploadedFiles
        this.chunksConcurrenceUploadNum = parseInt(String(10 / this.waitUploadFiles.length))
    }

    // 添加文件方法 （使用typescript写法）
    /**
     * @function 对外暴露的添加文件方法
     * @param newFiles 新添加的文件数组
     */
    public addNewFiles(newFiles: FileList) {

    }

    /**
     * @function 获取文件切片以及hash
     */
    private async calculateFilesMessage() {

    }


    /**
     * @function 添加已上传文件并上报
     * @param fileName 上传成功的文件名
     * @param url 返回的url
     */
    private addUploadedFiles(fileName: string, url: string): void {

    }

    /**
     * @function 增加计算hash完成文件并上报 调用上传方法
     * @param newWaitUploadFile 计算hash完成的文件
     */
    private async addCalculatedFile(newWaitUploadFile: IwaitUploadFile): Promise<any> {

    }

    /**
     * @function 执行验证以及上传逻辑 
     * @param waitUploadFile 待上传文件信息（内部的参数在接口中已经定义）
     */
    private async upload(waitUploadFile: IwaitUploadFile) {

    }

    /**
     * @function 计算已上传的size
     * @param AlreadyUploadList 服务端返回的已上传hash列表
     * @param waitUploadFile 待上传文件
     * @returns 已上传的切片大小
     */
    private calculeateAlreadyUploadSize(AlreadyUploadList: Array<any>, waitUploadFile: IwaitUploadFile) {

    }

    /**
     * @function 处理上传完成后的逻辑，上报更新UI
     * @param id 文件的id
     * @param fileName 文件名
     * @param url 得到的url
     */
    private completeFileUpload(id: string, fileName: string, url: string) {

    }

    /**
     * @function 发送上传切片前的的验证请求
     * @param fileName 文件名
     * @param filehash  文件hash值
     * @returns request对象
     */
    private verifyRequest(fileName: string, filehash: string) {

    }

    /**
     * 
     * @param id 待更改的文件id
     * @param e onprogress返回值
     * @param index 切片的下标
     * @function 更新上传进度
     */
    private updateUploadFilePercent(id: string, e: any, index: number): void {

    }

    /**
     * 
     * @param uploadFile 计算完成的文件
     * @function 发送文件合并请求
     */
    private mergeRequest(uploadFile: IwaitUploadFile) {

    }

}

具体的方法实现

到这里我们的文件大体结构就已经写好了，可以看到工具类对外只暴露添加文件方法，其他方法均作为工具方法供自己调用，?接下来我们就一个一个完善上面的方法

每一个方法都会有一定的注释，每个方法需要的依赖也会在方法上面进行标注

添加新文件方法

public addNewFiles(newFiles: FileList) {
           //  循环传入的FileList对象
        for (let i = 0, len = newFiles.length; i < len; i++) {
            this.waitCalculateFiles.push({
                id: `${newFiles[i].name}_${new Date().getTime()}`,
                file: newFiles[i],
            })
        }
        // 调用传入的更新文件函数上报
        this.updateWaitCalculateFile(this.waitCalculateFiles)
        // 切割已添加文件并开始计算已添加的文件的hash
        this.calculateFilesMessage()
  }

文件切片并计算文件hash方法(在类中编写)

逻辑： 每次从待上传文件队列中取出队首元素，对文件进行切片，切片完成后利用切片数组计算hash

依赖

    /**
     * @function 获取文件切片以及hash
     */
    import getFileChunkList from './utils/getFileChunkHash'
    import { calculatehash } from './utils/createHash'

    private async calculateFilesMessage() {
        while (this.waitCalculateFiles.length > 0) {
            const file: any = this.waitCalculateFiles[0].file
            const waituploadFile: IwaitUploadFile = {
                id: `${file.name as String}_${new Date().getTime()}`,
                file: file,
                // 切片文件
                chunkList: getFileChunkList(file, this.chunkSize),
                uploadProcess: 0,
                uploadPercentArr: [],
                uploadedSize: 0,
            }
            // 同步的计算文件hash
            const hash: string = await calculatehash(waituploadFile.chunkList) as string
            waituploadFile.hash = hash
            console.log(hash)
            // hash计算完成，更新待计算文件数组
            this.waitCalculateFiles.shift()
            waituploadFile.chunkList.forEach((item: chunkListsFile, index: number) => {
                item.hash = `${hash}_${index}`
                item.index = index
                item.fileName = hash
            })

            // 初始化上传进度数组
            waituploadFile.uploadPercentArr = new Array(waituploadFile.chunkList.length).fill(0)

            // 更新计算完成文件数组
            this.addCalculatedFile(waituploadFile)
            // 上报新的待计算文件数组
            this.updateWaitCalculateFile(this.waitCalculateFiles)
        }
    }

上传方法（重点）

断点续传如何实现 & 如何计算上传进度

断点续传的核心思想就是利用服务端返回的已上传切片列表，首先计算出已上传文件（loaded）的大小，然后在待上传文件列表中过滤掉已上传文件，之后每次计算文件上传进度时将上传文件列表上传的文件大小累加后再加上**已上传文件（loaded）**即可得到总体的已上传大小，除以文件整体size就可以得到上传进度

逻辑： 首先将该文件的信息发送至服务端，判断该文件是否需要上传，若存在直接调用文件上传完方法并返回，若服务端没有该文件，则判断服务端是否存在该文件切片，若存在则进行过滤等操作，然后开始并发的上传切片

依赖

/**
* @function 执行验证以及上传逻辑 
* @param waitUploadFile 待上传文件
*/
private async upload(waitUploadFile: IwaitUploadFile) {
    // 获取验证信息, 判断文件是否上传，以及已上传文件的信息，处理断点续传
    let verifyData: any = await this.verifyRequest(waitUploadFile.file.name, waitUploadFile.hash as string)
    verifyData = JSON.parse(verifyData.data)


    // 文件已经上传完成
    if (verifyData.status === 1) {
        // 若文件已经上传则调用上传完成方法，并传入相应的信息
        this.completeFileUpload(waitUploadFile.id as string, waitUploadFile.file.name, verifyData.url as string)
        return
    }
    // 处理断点续传逻辑
    if(verifyData.AlreadyUploadList) {
        // 计算已上传文件大小
        let loaded = this.calculeateAlreadyUploadSize(verifyData.AlreadyUploadList, waitUploadFile)
        let index = getUploadingFileIndexById(waitUploadFile.id as string, this.waitUploadFiles)
        if(index === -1) {
            return
        }
        this.waitUploadFiles[index].uploadedSize = loaded
        // 过滤已上传切片
        this.waitUploadFiles[index].chunkList = this.waitUploadFiles[index].chunkList.filter((item: chunkListsFile) => (
            verifyData.AlreadyUploadList.indexOf(item.hash) === -1
        ))
    }
    // 开始上传
    uploadFile(waitUploadFile, this.concurrency, this.updateUploadFilePercent.bind(this)).then(async res => {
        let uploadedMessage: any = await this.mergeRequest(waitUploadFile)
        uploadedMessage = JSON.parse(uploadedMessage.data)
        this.completeFileUpload(waitUploadFile.id as string, waitUploadFile.file.name, uploadedMessage.url as string)
    })
}

该方法内部uploadFile需要单独说一下

uploadFile并发上传切片方法

思路： 利用浏览器可以开启多个(chrome下6个)http请求线程进行并发上传，即js异步的开启，浏览器并行上传

import { IwaitUploadFile, chunkListsFile } from "../interfaces/interfaces";
import servicePath from './Apiurl'

/**
 * @param file 待上传文件
 * @param concurrency 并发上传最大数
 * @param callback 回调函数，上报修改上传进度
 * @function 上传文件
 * @returns promise对象，若上传完成则调用
 */

export default async function UploadFile(file: IwaitUploadFile, concurrency: number, updatePercent: (id: string, e: any, index: number) => void) {
    return new Promise((reslove, reject) => {
        let chunkList: Array<chunkListsFile> = file.chunkList
        let len = chunkList.length
        let counter = 0
        let isStop = false
        const start = async () => {
            if (isStop) {
                return
            }
            const item: chunkListsFile = chunkList.shift() as chunkListsFile
            if (item) {
                const formdata = new FormData()
                formdata.append('fileData', item.file)
                formdata.append('fileName', item.fileName)
                formdata.append('hash', item.hash)
                const xhr = new XMLHttpRequest()
                const index:number = item.index as number

                xhr.onerror = function error(e) {
                    isStop = true
                    reject(e)
                }

                // 分片上传完后的回调
                xhr.onload = () => {
                    if (xhr.status < 200 || xhr.status >= 300) {
                        isStop = true
                        reject('服务端返回状态码错误')
                    }
                    // 最后一个切片已经上传完成
                    if (counter === len - 1) {
                        reslove()
                    } else {
                        counter++
                        // 递归调用
                        start()
                    }
                }

                // 上报分片上传进度
                if (xhr.upload) {
                    xhr.upload.onprogress = (e: any) => {
                        if (e.total > 0) {
                            e.percent = e.loaded / e.total * 100
                        }
                        updatePercent(file.id as string, e, index)
                    }
                }
                xhr.open('post', servicePath.sendChunkRequest, true)
                xhr.send(formdata)
            }
        }
        // 开启并发数个函数进行递归请求
        while (concurrency > 0) {
            setTimeout(() => {
                start()
            , Math.random() * 100)
            concurrency--
        }
    })
}

计算更新文件上传进度的方法

逻辑
  • 得到当前文件下标fileIndex
  • 更新该文件上传进度列表
  • 计算该文件上传进度
  • 调用上层传入的函数上报文件最近上传进度更新UI
依赖
Import { IwaitUploadFile } from "../interfaces/interfaces"

export default function calculateUploadProcess(uploadedSize: number, waitUploadFile: IwaitUploadFile): number {
    // 根据服务端返回的已上传切片列表计算得到
    let loaded = uploadedSize
    // 遍历当前上传列表，得到已上传总进度
    waitUploadFile.uploadPercentArr.forEach((item: number) => {
        loaded += item
    })
    return loaded / waitUploadFile.file.size
}

/**
 * 
 * @param id 待更改的文件id
 * @param e onprogress返回值
 * @param index 切片的下标
 * @function 更新上传进度
 */
private updateUploadFilePercent(id: string, e: any, index: number): void {
    const fileIndex: number = getUploadingFileIndexById(id, this.waitUploadFiles)
    if (fileIndex === -1) { return }
    this.waitUploadFiles[fileIndex].uploadPercentArr[index] = e.loaded
    this.waitUploadFiles[fileIndex].uploadProcess =         calculateUploadProcess(this.waitUploadFiles[fileIndex].uploadedSize, this.waitUploadFiles[fileIndex])
    this.updateWaitUploadFile(this.waitUploadFiles)
}

后端

首先梳理一下后端我们需要做什么？

  • 接受文件切片并存储，收到文件合并请求的时候合并相应的文件
  • 验证文件是否存在
  • 返回已上传文件切片列表

接受文件切片存储

var express = require('express');
const path = require('path')
var app = express();
var fs = require('fs');
app.use('*', function (req, res, next) {
    res.header("Access-Control-Allow-Methods", "PUT,POST,GET,DELETE,OPTIONS");
    res.header("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept");
    res.setHeader('Access-Control-Allow-Credentials', true);
    res.setHeader('Access-Control-Allow-Origin', '*');
    next(); // 链式操作
});
//引入中间件
const ip = '127.0.0.1'
var formidable = require('formidable');
app.use(express.static('bigfiles'))
//监听路由


// 实现将文件切片存储到对应的文件夹下
// 实现的功能：

// 在uploads文件夹下，将当个切片存储为fileName为文件夹，hash_name+文件后缀为文件名
// hash_name = hash + 文件切片的下标
app.post('/file_upload', (req, res) => {
    console.log('接收到请求')
    //创建实例
    var form = new formidable.IncomingForm();
    //设置上传文件存放的目录
    form.uploadDir = "./uploads";
    //保持原来的文件的扩展名
    form.keepExtensions = true;
    //解析表单（异步）
    form.parse(req, function (err, fields, files) {
        //打印普通参数
        if (err) {
            console.log(`解析失败 ${err}`)
            return
        }
        console.log(fields);
        // 将文件名作为文件夹名，每个切片的hash_name作为切片文件名存储在对应文件夹下
        fs.mkdir(`${__dirname}/uploads/${fields.fileName}`, { recursive: true }, err => {
            if (err) {
                console.log('创建文件夹出错:' + err)
            } else {
                console.log(files);
                let oldPath = __dirname + '/' + files.fileData.path;
                let newPath = `${__dirname}/uploads/${fields.fileName}/${fields.hash}`;
                fs.rename(oldPath, newPath, err => {
                    if (err) { console.log(err) }
                })
            }
            res.send('ok');
            res.end();
        })
        //打印当前文件信息
    });
})

接受合并请求并合并对应文件

逻辑：

  1. 获取filename文件夹下的文件数组chunksDir
  2. 在bigfiles文件夹下建立一个文件hash+后缀的文件
  3. 遍历待合文件数组，以异步管道流的方式将单个文件输送至目标文件指定位置，整体使用promise.all进行监控
app.post('/mergeReq', async (req, res) => {
    console.log('====> 收到合并请求');
    const data = await reslovePost(req);
    await mergeFileChunks(data)
    res.send(JSON.stringify({
        ok: 1,
        msg: '合并完成',
        url: `http://${ip}:8001/${data.newname}`
    }))
    res.end()
})

const mergeFileChunks = async data => {
    const chunksDir = __dirname + `/uploads/${data.filename}`
    fs.readdir(chunksDir, async (err, files) => {
        if (err) {
            console.log(err)
            return
        }
        const chunkFilesPath = files.map(item => `${chunksDir}/${item}`)
        chunkFilesPath.sort(compareFun)
        await Promise.all(
            /**
             * 异步的将每一个文件item写入创建的文件可写流里
             */
            chunkFilesPath.map((item, index) =>
                pipeStream(
                    item,
                    fs.createWriteStream(`${__dirname}/bigfiles/${data.newname}`, {
                        flag: 'a+',
                        start: index * data.chunkSize,
                        end: (index + 1) * data.chunkSize > data.size ? data.size : (index + 1) * data.chunkSize
                    })
                )
            )
        )

        fs.rmdirSync(chunksDir, { recursive: true }, err => {
            console.log(chunksDir)
            console.log(err)
        })
    })
}

const pipeStream = (item, wirteStream) => {
    return new Promise(reslove => {
        const readStream = fs.createReadStream(item)
        readStream.on('end', () => {
            fs.unlinkSync(item, err => console.log(err))
            reslove()
        })
        readStream.pipe(wirteStream)
    })
}

验证请求

逻辑： 首先在bigfiles判断该文件是否存在，若存在则返回存在，并返回对应的文件信息，若不存在则在upload文件夹中将该文件对应的文件夹下的切片文件名信息返回给前端

app.post('/verify', async (req, res) => {
    console.log('=====>  收到验证请求')
    const data = await reslovePost(req)
    const { fileName, fileHash } = data
    const ext = getFileExt(fileName)
    const filePath = `${__dirname}/bigfiles/${fileHash}.${ext}`
    const hashPath = `${__dirname}/uploads/${fileHash}`
    if (fs.existsSync(filePath)) {
        console.log('=====> 该文件存在')
        res.send(JSON.stringify({
            status: 1,
            msg: '文件存在',
            isUpload: true,
            url: `http://${ip}:8001/${fileHash}.${ext}`
        }))
        res.end()
    } else {
        console.log('=====> 文件不存在')

        // 得到已上传切片列表
        const AlreadyUploadList = await getAlreadyUploadList(hashPath)
        console.log(AlreadyUploadList)
        res.send(JSON.stringify({
            status: 0,
            msg: '文件不存在',
            isUpload: false,
            AlreadyUploadList: AlreadyUploadList
        }))
        res.end()
    }
})
// 得到已上传切片文件列表
const getAlreadyUploadList = hashPath => {
    console.log(hashPath)
    return new Promise(reslove => {
        if (fs.existsSync(hashPath)) {
            fs.readdir(hashPath, (err, data) => {
                if (err) {
                    console.log(err)
                    reslove([])
                }
                reslove(data)
            })
        } else {
            reslove([])
        }
    })

}

总结

该工具的实现从逻辑上来说不是特别复杂，难点在于如何多个切片并发的同时计算出文件的上传进度以及如何控制同时并发上传数量

相关资讯:

  1. IE8兼容处理之上传文件stream-v1.js
  2. 大文件上传vue+WebUploader
  3. 多文件上传你学废了吗？
  4. 前端的文件流-flie介绍
标签：, ,
声明：本文内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎进行举报，并提供相关证据，工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。

今日推荐

Python小白的数学建模课-B5. 新冠疫情 SEIR模型

传染病的数学模型是数学建模中的典型问题，常见的传染病模型有 SI、SIR、SIRS、SEIR 模型。 考虑存在 […]