从 Node 上传服务说起的爬坑日记

[rottenpen]

背景

上传接口需要上传 zip 文件的同时对 zip 文件进行解压上传。接下来我会介绍一下整个上传服务的优化过程。

Step

单文件上传

首先我们来看一下，单文件上传，是怎么实现的（因为 sketch 插件的 formdata 是三方开发的，这篇指南中没有用到任何 formdata 的 request payload）。

const app = express();
app.post("/upload", (req, res) => {
  // 因为 body 都用来放文件的 bufferArray
	const fileName = req.query.fileName;
  const TEMP_DIR = path.resolve(__dirname, "..", "tmp");
  if (!fileName) {
    res.status(400).json({ c: "-1", m: "query 缺少 fileName!" });
    return;
  }
	const chunkDir = path.resolve(TEMP_DIR);
  const prefix = +new Date();
  const fn = `${chunkDir}/${prefix}/${fileName}`;
  // fs-extra 提供的确认文件夹有没有生成
  fs.ensureDirSync(chunkDir);
  fs.ensureDirSync(`${chunkDir}/${prefix}`);
  // 生成一个可写流 fn是可写流的写入路径
  const stream = fs.createWriteStream(fn);
  // 通过pipe 把整个 request body 写入可写流
  const pipfile = req.pipe(stream).on("error", () => {
    res.status(400).json({ c: "-1", m: "上传失败,接收文件失败" });
  });
  // close 的时候，可写流结束，文件成功生成了
  pipfile.on("close", () => {
    try {
			// 上传到 cos，里面如果上传成功就通过 
      // res.send({ c: "0", d: response, m: "ok" }) 返回结果
      upload(res); 
    } catch (error) {
      res.status(400).json({ c: "-1", m: error });
    }
  });
});

上面就是一个单文件上传接口的实现，简单总结一下它都做了什么：

1. 创建可写流
2. 把 request pipe 进可写流
3. 通过 cos sdk 上传文件

这种实现有什么问题呢？

上传大文件时间太长，有可能造成超时。

分片上传

然后我改成了分片上传，分片上传不是

worker_threader

即使我们使用了分片上传，我们依然没法解决接口阻塞问题，在上传过程中，有新的请求接入，需要等上传的 callback 完成之后。我一开始想到的是，难道网络请求的异步 I/O 也能阻塞整个进程吗？于是我把上传的部分放到 worker 上运行，效果显著，马上不阻塞了。

但是明明上传操作应该是异步请求，为什么会阻塞呢？

看了一下，原来是在读取文件信息的时候，我们封装的 cos sdk 用到了很多同步操作。

当文件数很多的时候就会阻塞我们的请求了。原来凶手在这⬆️ （ps. 除了同步 fs 还有很多打 log 的操作，console.log 也是一个同步阻塞的 io 操作，从下图可以看出，log 的性能不比 readfile 好多少）

EventLoop

上面我们分析了，一个上传接口怎么写，如何进行分片，如何处理 I/O 密集型任务。本来故事到这里应该就完，但是我在写这篇分享的时候，一时之间陷入了迷思，为什么异步 I/O 会阻塞请求新请求的回调呢？接收请求的回调是微任务还是宏任务呢？

先做一个小实验。

const { EventEmitter } = require("events");
let ee = new EventEmitter();
ee.on("log", console.log);
for (let i = 0; i < 1000; i++) {
  Promise.resolve().then(() => console.log(i))
  if ((i === 50)) {
    ee.emit("log", `---------${i}`);
  }
}

这段代码的结果是， ——--50 比 console.log(i) 更早打印出来。其实如果有自己手写过 eventEmitter 的同学都知道，其实 emit 方法只是调用以下哈希表里对应的方法而已，这个方法是同步的，因为 node 的 http 模块是基于 event 的，express 的 listen 也是基于 event 的，我们是不是就可以得出他们的请求也是同步的结论呢。

OK，下面我们跳出日常浏览器宏任务微任务的八股。来看看 node 环境下的 Event loop (以下出自 node 官方文档: https://nodejs.org/zh-cn/docs/guides/event-loop-timers-and-nexttick/):
当Node.js启动时会初始化event loop, 每一个event loop都会包含按如下顺序六个循环阶段：

阶段概述

• 定时器：本阶段执行已经被 setTimeout() 和 setInterval() 的调度回调函数。
• 待定回调：执行延迟到下一个循环迭代的 I/O 回调。
• idle, prepare：仅系统内部使用。
• 轮询：检索新的 I/O 事件;执行与 I/O 相关的回调（几乎所有情况下，除了关闭的回调函数，那些由计时器和 setImmediate() 调度的之外），其余情况 node 将在适当的时候在此阻塞。
• 检测：setImmediate() 回调函数在这里执行。
• 关闭的回调函数：一些关闭的回调函数，如：socket.on('close', ...)。

我们通过两段异步操作的代码来深入我们对 node eventLoop 的理解。

setTimeout(() => {
  console.log('timer1');
  Promise.resolve().then(function() {
    console.log('promise1');
  });
}, 0);
setTimeout(() => {
  console.log('timer2');
  Promise.resolve().then(function() {
    console.log('promise2');
  });
}, 0);

OK，我们再来看看下一段代码

const fs = require("fs");
setTimeout(() => {
  console.log("timer1");
  fs.readFile(__dirname + "/" + __filename, () => {
    console.log("fs1");
  });
}, 0);
setTimeout(() => {
  console.log("timer2");
  fs.readFile(__dirname + "/" + __filename, () => {
    console.log("fs2");
  });
}, 0);

第一段代码在不同版本的表现是不一样的。

在 node 10 中，第一段代码的结果是

timer1 timer2 promise1 promise2

因为 node 会优先把 timer 队列清空再执行微任务的 nexttick 队列。

而在 node 11 后，我们得到的结果是。

timer1 promise1 timer2 promise2

仔细去翻node的修改日志，在node 11.0 的修改日志里面发现了这个：

• Timers
  • nextTick queue will be run after each immediate and timer. #22842

也就是说 nextick 队列会在每一个 timer 和 immediate 后执行。也就是说微任务队列会在，每一次执行 settimeout , setImmediate , setInterval 的 callback 后执行。

如何实现的

下面我们来看看他是怎么实现的 ( 参考PR https://github.com/nodejs/node/pull/22842/files#diff-5a0457600721c223f1ed7184ef7d1d2617f4552a5341b53a49b284f808981724)

这是一个遍历 timer 队列的函数，当 while 循环执行了一次之后，ranAtLeastOneTimer 会变为 true ，然后执行 runNextTicks() 即立即执行微任务队列。

如果我们想在 node 10 中也能看到类似的效果，我们可以：

setTimeout(() => {
    console.log('timer1');
    Promise.resolve().then(function() {
        console.log('promise1');
    });
    process._tickCallback(); // 这行是增加的！
}, 0);
setTimeout(() => {
    console.log('timer2');
    Promise.resolve().then(function() {
        console.log('promise2');
    });
    process._tickCallback(); // 这行是增加的！
}, 0);

OK，接下来就是垃圾时间了我们来看看第二段代码的结果。

timer1 timer2 fs1 fs2

fs1 fs2 是在 poll 阶段执行的，它们在执行完 timer 之后需要一段时间的异步 I / O 才能被执行。

总结

所以回到上面我的迷思，网络请求的异步回调虽然是基于 event 机制实现的，但它其实是在 poll 阶段被异步执行。发送过来的请求被阻塞是因为我之前打太多 log 以及 fs 同步调用被阻塞。Worker 在一定程度上解决了同步阻塞的问题，但生成线程的开销也不容忽视，在非阻塞的情况下使用 Worker 并不是一个很好的方案。