node+js实现大文件分片上传

Posted 2021-02-20 古兰精

1、什么是分片上传

　　分片上传就是把一个大的文件分成若干块，一块一块的传输。这样做的好处可以减少重新上传的开销。比如：如果我们上传的文件是一个很大的文件，那么上传的时间应该会比较久，再加上网络不稳定各种因素的影响，很容易导致传输中断，用户除了重新上传文件外没有其他的办法，但是我们可以使用分片上传来解决这个问题。通过分片上传技术，如果网络传输中断，我们重新选择文件只需要传剩余的分片。而不需要重传整个文件，大大减少了重传的开销。

　　但是我们要如何选择一个合适的分片呢？因此我们要考虑如下几个事情：

　　1. 分片越小，那么请求肯定越多，开销就越大。因此不能设置太小。
　　2. 分片越大，灵活度就少了。
　　3. 服务器端都会有个固定大小的接收Buffer。分片的大小最好是这个值的整数倍。

　　因此，综合考虑到推荐分片的大小是2M-5M，具体分片的大小需要根据文件的大小来确定，如果文件太大，建议分片的大小是5M，如果文件相对较小，那么建议分片的大小是2M。

　　实现文件分片上传的步骤如下：

　　1. 先对文件进行md5加密。使用md5加密的优点是：可以对文件进行唯一标识，同样可以为后台进行文件完整性校验进行比对。
　　2. 拿到md5值以后，服务器端查询下该文件是否已经上传过，如果已经上传过的话，就不用重新再上传。
　　3. 对大文件进行分片。比如一个100M的文件，我们一个分片是5M的话，那么这个文件可以分20次上传。
　　4. 向后台请求接口，接口里的数据就是我们已经上传过的文件块。(注意：为什么要发这个请求？就是为了能断点续传，比如我们使用百度网盘对吧，网盘里面有续传功能，当一个文件传到一半的时候，突然想下班不想上传了，那么服务器就应该记住我之前上传过的文件块，当我打开电脑重新上传的时候，那么它应该跳过我之前已经上传的文件块。再上传后续的块)。
　　5. 开始对未上传过的文件块进行上传。(这个是第二个请求，会把所有的分片合并，然后上传请求)。
　　6. 上传成功后，服务器会进行文件合并。最后完成。

2、理解Blob对象中的slice方法对文件进行分割及其他知识点

　　可以看下我之前的博客：利用blob对象实现大文件分片上传

　　Blob对象自身有 size 和 type两个属性，及它的原型上有 slice() 方法。我们可以通过该方法来切割我们的二进制的Blob对象。

　　blob.slice(startByte, endByte) 是Blob对象中的一个方法，File对象它是继承Blob对象的，因此File对象也有该slice方法的。

　　参数：
　　　　startByte: 表示文件起始读取的Byte字节数。
　　　　endByte: 表示结束读取的字节数。

　　返回值：var b = new Blob(startByte, endByte); 该方法的返回值仍然是一个Blob类型。

　　我们可以使用 blob.slice() 方法对二进制的Blob对象进行切割，但是该方法也是有浏览器兼容性的，因此我们可以封装一个方法：如下所示：

function blobSlice(blob, startByte, endByte) {
  if (blob.slice) {
    return blob.slice(startByte, endByte);
  }
  // 兼容firefox
  if (blob.mozSlice) {
    return blob.mozSlice(startByte, endByte);
  }
  // 兼容webkit
  if (blob.webkitSlice) {
    return blob.webkitSlice(startByte, endByte);
  }
  return null;
}

3、具体实现

$(document).ready(() => {
  const chunkSize = 2 * 1024 * 1024; // 每个chunk的大小，设置为2兆
  // 使用Blob.slice方法来对文件进行分割。
  // 同时该方法在不同的浏览器使用方式不同。
  const blobSlice = File.prototype.slice || File.prototype.mozSlice || File.prototype.webkitSlice;
  const hashFile = (file) => {
    return new Promise((resolve, reject) => { 
      const chunks = Math.ceil(file.size / chunkSize);
      let currentChunk = 0;
      const spark = new SparkMD5.ArrayBuffer();
      const fileReader = new FileReader();
      function loadNext() {
        const start = currentChunk * chunkSize;
        const end = start + chunkSize >= file.size ? file.size : start + chunkSize;
        fileReader.readAsArrayBuffer(blobSlice.call(file, start, end));
      }
      fileReader.onload = e => {
        spark.append(e.target.result); // Append array buffer
        currentChunk += 1;
        if (currentChunk < chunks) {
          loadNext();
        } else {
          console.log(\'finished loading\');
          const result = spark.end();
          // 如果单纯的使用result 作为hash值的时候, 如果文件内容相同，而名称不同的时候
          // 想保留两个文件无法保留。所以把文件名称加上。
          const sparkMd5 = new SparkMD5();
          sparkMd5.append(result);
          sparkMd5.append(file.name);
          const hexHash = sparkMd5.end();
          resolve(hexHash);
        }
      };
      fileReader.onerror = () => {
        console.warn(\'文件读取失败！\');
      };
      loadNext();
    }).catch(err => {
        console.log(err);
    });
  }
  const submitBtn = $(\'#submitBtn\');
  submitBtn.on(\'click\', async () => {
    const fileDom = $(\'#file\')[0];
    // 获取到的files为一个File对象数组，如果允许多选的时候，文件为多个
    const files = fileDom.files;
    const file = files[0];
    if (!file) {
      alert(\'没有获取文件\');
      return;
    }
    const blockCount = Math.ceil(file.size / chunkSize); // 分片总数
    const axiosPromiseArray = []; // axiosPromise数组
    const hash = await hashFile(file); //文件 hash 
    // 获取文件hash之后，如果需要做断点续传，可以根据hash值去后台进行校验。
    // 看看是否已经上传过该文件，并且是否已经传送完成以及已经上传的切片。
    console.log(hash);
    
    for (let i = 0; i < blockCount; i++) {
      const start = i * chunkSize;
      const end = Math.min(file.size, start + chunkSize);
      // 构建表单
      const form = new FormData();
      form.append(\'file\', blobSlice.call(file, start, end));
      form.append(\'name\', file.name);
      form.append(\'total\', blockCount);
      form.append(\'index\', i);
      form.append(\'size\', file.size);
      form.append(\'hash\', hash);
      // ajax提交 分片，此时 content-type 为 multipart/form-data
      const axiosOptions = {
        onUploadProgress: e => {
          // 处理上传的进度
          console.log(blockCount, i, e, file);
        },
      };
      // 加入到 Promise 数组中
      axiosPromiseArray.push(axios.post(\'/file/upload\', form, axiosOptions));
    }
    // 所有分片上传后，请求合并分片文件
    await axios.all(axiosPromiseArray).then(() => {
      // 合并chunks
      const data = {
        size: file.size,
        name: file.name,
        total: blockCount,
        hash
      };
      axios.post(\'/file/merge_chunks\', data).then(res => {
        console.log(\'上传成功\');
        console.log(res.data, file);
        alert(\'上传成功\');
      }).catch(err => {
        console.log(err);
      });
    });
  });
})

　　我们需要获取分片的总数 —— 然后使用 for循环遍历分片的总数 —— 然后依次实例化formData数据 —— 依次把对应的分片添加到 formData数据里面去。

　　然后分别使用 \'/file/upload\' 请求数据，最后把所有请求成功的数据放入到 axiosPromiseArray 数组中，当所有的分片上传完成后，我们会使用 await axios.all(axiosPromiseArray).then(() => {}) 方法，最后我们会使用 \'/file/merge_chunks\' 方法来合并文件。

const Koa = require(\'koa\');
const app = new Koa();
const Router = require(\'koa-router\');
const multer = require(\'koa-multer\');
const serve = require(\'koa-static\');
const path = require(\'path\');
const fs = require(\'fs-extra\');
const koaBody = require(\'koa-body\');
const { mkdirsSync } = require(\'./utils/dir\');
const uploadPath = path.join(__dirname, \'uploads\');
const uploadTempPath = path.join(uploadPath, \'temp\');
const upload = multer({ dest: uploadTempPath });
const router = new Router();
app.use(koaBody());
/**
 * single(fieldname)
 * Accept a single file with the name fieldname. The single file will be stored in req.file.
 */
router.post(\'/file/upload\', upload.single(\'file\'), async (ctx, next) => {
  console.log(\'file upload...\')
  // 根据文件hash创建文件夹，把默认上传的文件移动当前hash文件夹下。方便后续文件合并。
  const {
    name,
    total,
    index,
    size,
    hash
  } = ctx.req.body;

  const chunksPath = path.join(uploadPath, hash, \'/\');
  if(!fs.existsSync(chunksPath)) mkdirsSync(chunksPath);
  fs.renameSync(ctx.req.file.path, chunksPath + hash + \'-\' + index);
  ctx.status = 200;
  ctx.res.end(\'Success\');
})

router.post(\'/file/merge_chunks\', async (ctx, next) => {
  const {    
    size, 
    name, 
    total, 
    hash
  } = ctx.request.body;
  // 根据hash值，获取分片文件。
  // 创建存储文件
  // 合并
  const chunksPath = path.join(uploadPath, hash, \'/\');
  const filePath = path.join(uploadPath, name);
  // 读取所有的chunks 文件名存放在数组中
  const chunks = fs.readdirSync(chunksPath);
  // 创建存储文件
  fs.writeFileSync(filePath, \'\'); 
  if(chunks.length !== total || chunks.length === 0) {
    ctx.status = 200;
    ctx.res.end(\'切片文件数量不符合\');
    return;
  }
  for (let i = 0; i < total; i++) {
    // 追加写入到文件中
    fs.appendFileSync(filePath, fs.readFileSync(chunksPath + hash + \'-\' +i));
    // 删除本次使用的chunk    
    fs.unlinkSync(chunksPath + hash + \'-\' +i);
  }
  fs.rmdirSync(chunksPath);
  // 文件合并成功，可以把文件信息进行入库。
  ctx.status = 200;
  ctx.res.end(\'合并成功\');
})
app.use(router.routes());
app.use(router.allowedMethods());
app.use(serve(__dirname + \'/static\'));
app.listen(9000, () => {
  console.log(\'服务9000端口已经启动了\');
});

　　utils/dir.js，该代码的作用是判断是否有这个目录，有这个目录的话，直接返回true，否则的话，创建该目录

const path = require(\'path\');
const fs = require(\'fs-extra\');
const mkdirsSync = (dirname) => {
  if(fs.existsSync(dirname)) {
    return true;
  } else {
    if (mkdirsSync(path.dirname(dirname))) {
      fs.mkdirSync(dirname);
      return true;
    }
  }
}
module.exports = {
  mkdirsSync
};

　　我们先看 \'/file/upload\' 这个请求，获取到文件后，请求成功回调，然后会在项目中的根目录下创建一个 uploads 这个目录

　　我们也可以在我们的网络中看到很多 \'/file/upload\' 的请求，说明我们的请求是分片上传的

　　最后所有的分片请求上传成功后，我们会调用 \'/file/merge_chunks\' 这个请求来合并所有的文件，根据我们的hash值，来获取文件分片。然后我们会循环分片的总数，然后把所有的分片写入到我们的filePath目录中

fs.appendFileSync(filePath, fs.readFileSync(chunksPath + hash + \'-\' +i));

　　其中 filePath 的获取 是这句代码：const filePath = path.join(uploadPath, name); 也就是说在我们项目的根目录下的uploads文件夹下，这么做的原因是为了防止网络突然断开或服务器突然异常的情况下，文件上传到一半的时候，我们本地会保存一部分已经上传的文件，如果我们继续上传的时候，我们会跳过哪些已经上传后的文件，继续上传未上传的文件。这是为了断点续传做好准备的，下次我会分析下如何实现断点续传的原理了。

　　如上就是我们整个分片上传的基本原理，我们还没有做断点续传了，下次有空我们来分析下断点续传的基本原理，断点续传的原理，无非就是说在我们上传的过程中，如果网络中断或服务器中断的情况下，我们需要把文件保存到本地，然后当网络恢复的时候，我们继续上传，那么继续上传的时候，我们会比较上传的hash值是否在我本地的hash值是否相同，如果相同的话，直接跳过该分片上传，继续下一个分片上传，依次类推来进行判断，虽然使用这种方式来进行比对的情况下，会需要一点时间，但是相对于我们重新上传消耗的时间来讲，这些时间不算什么的。下次有空我们来分析下断点续传的基本原理哦。