大文件上传？其实真的没有那么难（二）

作为一名前端练习生，经常看到有关大文件上传的文章或视频，但从未动手去实现过，今天终于鼓起勇气，认真的分析了一下，发现其实并没有那么复杂。

源码地址 (kakachake/BigFile: 大文件上传示例 (github.com))

一、前言

本文是大文件上传？其实真的没有那么难！（一）的第二篇文章，在上一篇文章中，我们已经完成了对BigFile的封装，通过他我们可以对用户上传的文件进行切片了，接下来要做的就是把切片文件发送给服务端并通知合并了，话不多说，我们开始吧！

二、实现

1. 前端

（1）上传切片

到现在为止，我们已经完成了对文件的切片，那么接下来我们就应该上传切片了，我们首先在前端页面定义一个uploadChunks函数吧：

// app.tsx
const uploadChunks = async function (
    chunks: IFileChunk[],
    fileName: string,
    hash: string
) {
    //……
});

首先我们要做下数据处理：

这里我们思考下，上传切片，我们要做什么？或者说，服务端需要什么？

很容易可以想到，服务端如果要合并分片文件，必须知道当前分片属于哪个文件，其次还需要知道当前这个分片文件的分片序号，这样才能进行分片文件的合并。

这里我们将分片文件的hash值设置成源文件的hash加上index,那么我们传递给服务端的内容就很清晰了：

• chunk：分片文件
• hash：分片文件hash (源文件hash+下标)
• filename：源文件名

我们可以创建一个createFormData来做这项工作：

const createFormData = function (data: IFileChunk, fileName: string) {
    const formData = new FormData();
    formData.append("chunk", data.chunk);
    formData.append("hash", data.hash);
    formData.append("filename", fileName);
    return formData;
};

这里可以看到切片文件的hash是直接在data中取到的，而这个data实际上就是从我们上一篇讲到的BigFile的getChunks函数中返回的chunks数组的某一项，里面是包含hash字段的，如果不记得了，可以去看一下上一篇文章。

拿到处理好的formData数据，我们就可以上传分片了，uploadChunks的代码如下：

const uploadChunks = async function (
    chunks: IFileChunk[],
    fileName: string,
    hash: string
) {
    // controller.current = new AbortController();
    // 设置进度条
    const progresses: number[] = new Array(chunks.length).fill(0);
    setProgresses(progresses);
    // 处理数据
    const formDatas = chunks
      .map((chunk) => {
        return {
          formData: createFormData(chunk, fileName),
          index: chunk.index,
        };
      });

    // 创建请求函数数组
    const reqs = formDatas.map((item, index) => {
      return axios.post.bind(
        null,
        "http://localhost:3000/upload",
        item.formData,
        {
          onUploadProgress: (e) => {
            const total = e.total;
            const progress =
              (total && Math.round((e.loaded * 100) / total)) || 0;
            const idx = item.index;
            progresses[idx] = progress;
            setProgresses([...progresses]);
            if (progress === 100) {
              chunks[idx].complete = true;
            }
          },
          // signal: controller.current!.signal,
        }
      );
    });
    
    // 并行限制发送分片数据
    await multiRequest(reqs, 10);
    messageApi.open({
      type: "loading",
      content: "正在合并chunk文件..",
      duration: 0,
    });
    // 通知合并
    await mergeChunk(hash, fileName);
    messageApi.destroy();
    message.success("合并成功", 1);
};

这里可以看到我并没有直接for循环发送切片数据，而是返回了一个请求函数数组，这个数组保存了所有需要发送的请求函数，然后我编写了一个multiRequest函数，用于做并发限制：

// utils.ts
export const multiRequest = function (
  reqs: (() => Promise<any>)[],
  limit: number
): Promise<any[]> {
  return new Promise((resolve) => {
    const results: any[] = [];
    function* gen() {
      for (let i = 0; i < reqs.length; i++) {
        yield reqs[i]();
      }
    }
    const g = gen();
    function next() {
      const { value, done } = g.next();
      if (done) return;
      value.then(() => {
        results.push(value);
        if (results.length === reqs.length) {
          resolve(results);
        }
        next();
      });
    }
    for (let i = 0; i < limit; i++) {
      next();
    }
  });
};

有关promise并发限制的内容就不在这篇文章中详述了，大家可以关注我，未来我会专门出一篇文章去讲这个内容。

（2）合并切片

等待切片全部发送完毕后，就可以通知后端合并了：

const mergeChunk = async (hash: string, filename: string) => {
    await axios
      .post("http://localhost:3000/merge", {
        hash,
        filename,
      })
      .then((res) => {
        const { data } = res;
        if (data.code === 200) {
          // 合并成功！
        }
      });
};

这里我们给后端传递了文件的hash（用于查找分片文件）、filename（用于后端获取文件后缀）

前端的工作基本上到这里就结束了，下面我们看下后端的内容：

2. 后端

（1） 基本框架搭建

后端是使用的express搭建的，先装一下必要的依赖：

npm i express body-parser multiparty fs-extra

编写一些必要的代码：

// app.ts
import express from "express";
import bodyParser from "body-parser";
import multiparty from "multiparty";
import path from "path";

const app = express();

// 设置跨域
app.all("*", (req, res, next) => {
  res.setHeader("Access-Control-Allow-Origin", "*");
  res.setHeader("Access-Control-Allow-Headers", "*");
  next();
});

app.use("/files", express.static(path.resolve(UPLOAD_DIR, "file")));

// parse application/x-www-form-urlencoded
app.use(bodyParser.urlencoded({ extended: false }));

// parse application/json
app.use(bodyParser.json());

app.get("/", (req, res) => {
  res.send("大文件上传服务启动成功！");
});

app.listen(3000, () => {
  console.log("Server is running on port 3000");
});

这里使用到了multiparty来读取FormData的数据。

（2） 接收切片

首先编写接收切片的接口，接收切片的整个流程如下：

1. 接收切片数据
2. 解析chunk、hash、filename
3. 保存切片

具体代码为：

const getFileDir = (filename: string, hash: string) => {
  return "chunk-" + hash.slice(0, 5) + filename.split(".")[0];
};

app.post("/upload", (req, res) => {
  try {
    const form = new multiparty.Form({});
    form.parse(req, (err, fields, files) => {
      if (err) {
        console.log(err);
        return;
      }
      // 获取必要参数
      const [chunk] = files.chunk;
      const [hash] = fields.hash;
      const [filename] = fields.filename;
      
      // multiparty保存切片的临时位置
      const { path: oldPath } = chunk;
        
      // 默认保存切片的位置
      const chunkDir = path.resolve(
        UPLOAD_DIR,
        "temp",
        getFileDir(filename, hash)
      );
      
      // 创建文件夹
      if (!fse.existsSync(chunkDir)) {
        fse.mkdirSync(chunkDir);
      }
      
      // 将切片文件从临时位置移动到我们创建的路径下
      fse.moveSync(oldPath as string, `${chunkDir}/${hash}`, {
        overwrite: true,
      });
      
      res.status(200);
      res.setHeader("Content-Type", "application/json");
      res.send({ code: 200, msg: "上传成功", hash });
    });
  } catch (error) {
    res.status(500);
    res.setHeader("Content-Type", "application/json");
    res.send({ code: 500, msg: "上传失败", error });
  }
})

根据代码我们可以看到我们将接收到的文件都存放在了项目目录下的upload/temp/下，每个文件对应一个文件夹，文件夹名是chunk+hash[:5]+filename的组合，切片文件名为切片文件的hash。

（3）合并切片

接下来是合并的接口： 合并的流程如下：

1. 获取filename和hash
2. 根据filename和hash得到chunksdir、文件后缀
3. 遍历切片文件夹下的切片文件，合并切片文件

代码如下：

app.post("/merge", async (req, res) => {
  const { filename, hash } = req.body;
  // 获取切片文件路径
  const dirName = getFileDir(filename, hash);
  // 获取文件后缀
  const fileType = filename.split(".").pop();
  // 构造最终的文件名
  const mergeName = `${hash}.${fileType}`;
  const fileWriteStream = fs.createWriteStream(
    path.resolve(UPLOAD_DIR, "file", mergeName)
  );
  const chunksdir = path.resolve(UPLOAD_DIR, "temp", dirName);
  // 合并切片文件
  const mergeRes = await streamMerge(chunksdir, fileWriteStream);
  if (mergeRes.code === 0) {
    fse.removeSync(chunksdir);
  }
  res.status(200);
  res.setHeader("Content-Type", "application/json");
  res.send({ code: 200, msg: "合并成功", filename: mergeName });
});

// util.ts
async function asyncChunk(chunkStream: fs.ReadStream): Promise<{
  code: 0 | 1;
}> {
  return new Promise((resolve, reject) => {
    chunkStream.on("end", () => {
      resolve({
        code: 0,
      });
    });
    chunkStream.on("error", (err) => {
      reject({
        code: 1,
        msg: err,
      });
    });
  });
}

export const streamMerge = function (
  chunkDir: string,
  fileWriteStream: fs.WriteStream
) {
  // 获取切片文件，对切片文件按照序号从大到小排序，这里文件名为"hash_index"，故只需要取index排序即可
  const chunks = fs.readdirSync(chunkDir).sort((a, b) => {
    a = a.split("_").pop();
    b = b.split("_").pop();
    return Number(a) - Number(b);
  });

  return new Promise<{
    code: 0 | 1;
  }>(async (resolve, reject) => {
    // 遍历切片文件并合并
    while (chunks.length > 0) {
      const chunk = chunks.shift();
      const chunkPath = path.resolve(chunkDir, chunk);
      const chunkStream = fs.createReadStream(chunkPath);
      chunkStream.pipe(fileWriteStream, { end: false });
      // 将chunkStream转为promise异步形式
      const res = await asyncChunk(chunkStream);
      if (res.code === 1) {
        reject(res);
      }
    }
    resolve({
      code: 0,
    });
    fileWriteStream.end();
  });
};

三、最终效果

到此为止，整个大文件上传的核心逻辑就完成了，除此之外，还可以扩展暂停、恢复上传的功能，具体细节可以去github仓库查看。

最后运行测试一下，这里我设置的并发限制数量是10，正好留一个问题，为啥gif图里看起来并发限制是6呢？