Node.js 新特性:工作线程

目录

2019 年 09 月 26 日 node.js 12.11.0 发布,工作线程(多线程)稳定。

单进程和单线程

当一个 Node.js 的应用启动的同时,它会启动如下模块:

  • 一个进程
  • 一个线程
  • 事件循环机制
  • JS 引擎实例
  • Node.js 实例

一个进程:process 对象是一个全局变量,可在 Node.js 程序中任意地方访问,并提供当前进程的相关信息。
一个线程:单线程意味着在当前进程中同一时刻只有一个指令在执行。
事件循环:这是 Node.js 中需要重点理解的一个部分,尽管 JavaScript 是单线程的,但通过使用回调,promises, async/await 等语法,基于事件循环将对操作系统的操作异步化,使得 Node 拥有异步非阻塞 IO 的特性。
一个 JS 引擎实例:即一个可以运行 JavaScript 代码的程序。
一个 Node.js 实例:即一个可以运行 Node.js 环境的程序。

为什么需要多线程

Node.js 由于 JS 的执行在单一线程,导致 CPU 密集计算的任务可能会使主线程会处于繁忙的状态,进而影响服务的性能,虽然可以通过 child_process 模块创建子进程的方式来解决,
但是一方面进程之间无法共享内存,另一方面创建进程的开销也不小。所以在 10.5.0 版本中 Node.js 提供了 worker_threads 模块来支持多线程,一直以来被人所诟病的不擅长 CPU 密集计算有望成为历史。

多线程 Workers 下 Node.js 拥有:

  • 一个进程
  • 多个线程
  • 每个线程都拥有独立的事件循环
  • 每个线程都拥有一个 JS 引擎实例
  • 每个线程都拥有一个 Node.js 实例

Worker Threads

UTOOLS1569484023918.png

Worker Threads 有如下特性:

  • ArrayBuffers 可以将内存中的变量从一个线程转到另外一个
  • SharedArrayBuffer 可以在多个线程中共享内存中的变量,但是限制为二进制格式的数据。
  • 可用的原子操作,可以让你更有效率地同时执行某些操作并且实现竞态变量
  • 消息端口,用于多个线程间通信。可以用于多个线程间传输结构化的数据,内存空间
  • 消息通道就像多线程间的一个异步的双向通信通道。
  • WorkerData 是用于传输启动数据。在多个线程间使用 postMessgae 进行传输的时候,数据会被克隆,并将克隆的数据传输到线程的 contructor 中。

API:

const { worker, parantPort } = require("worker_threads");

worker 函数相当于一个独立的 JavaScript 运行环境线程,parentPort 是消息端口的一个实例

new Worker(filename);
//or
new Worker(code, { eval: true });

启动 worker 的时候有两种方式,可以通过传输文件路径或者代码,在生产环境中推荐使用文件路径的方式。

worker.on("message"), worker.postMessage(data);

这是多线程间监听事件与推送数据的方式。

parentPort.on("message");
parentPort.postMessage(data);

在线程中使用 parentPort.postMessage 方式推送的数据可以在父进程中使用 worker.on('message')
的方式接收到,在父进程中使用 worker.postMessage() 的方式推送的数据可以在线程中使用 parentPort.on('message') 的方式监听到。

什么是 ArrayBuffer

ArrayBuffer 跟其它 JavaScript 数组差不多,但是不是所有 JavaScript 类型都可以放进去,比如对象、字符串。你唯一可以放进去的只有字节。

ArrayBuffer 仅仅是一个个 0/1 组成的串,它不知道第一个元素和第二个元素的分割点。

为了提供必要的上下文信息,把 ArrayBuffer 分块,我们需要把它包裹到视图里,这些数据的视图可以通过带类型的数组添加,已经支持很多种类型的数组了

例如,你可以用一个 Int8 类型的数组把 0/1 串分割成 8 位一组的序列。

ArrayBuffer 几乎扮演原始内存角色,它模拟内存的各种跟 C 语言里类似的操作。

如何创建多线程

worker_threads 模块中比较重要的几个类:
MessageChannel 用于创建异步、双向通信的通道实例。MessageChannel 实例包含两个属性 port1port2,这两个属性都是 MessagePort 的实例。

MessagePort 用于表示 MessageChannel 通道的终端,用于 Worker 之间传输结构化数据、内存区域和其他的 MessagePortMessagePort 继承了 EventEmitter,因此可以使用 postMessageon 方法实现消息的传递与接收。

Worker: 用于创建单独的 JS 线程。

worker_threads 模块中比较重要的几个属性:
parentPort 子线程中的 parentPort 指向可以与主线程进行通信的 MessagePort

子线程向父线程发送消息

parentPort.postMessage(/*...*/);

子线程接受来自父线程的消息

parentPort.on("message", msg => {
  /*...*/
});

isMainThread 用于区分当前文件是否在主线程中执行

workerData 用于传递给 Worker 构造函数的 data 副本,在子线程中可以通过 workerData 获取到父进程传入的数据。

const { Worker, parentPort, isMainThread } = require("worker_threads");
if (isMainThread) {
  const w = new Worker(__filename, {
    workerData: {
      name: "Randal",
    },
  });
  w.postMessage(1e10);
  const startTime = Date.now();
  w.on("message", function(msg) {
    console.log("main thread get message: " + msg);
    console.log("compute time ellapsed: " + (Date.now() - startTime) / 1000);
  });
  console.log("main thread executing");
} else {
  const longComputation = val => {
    let sum = 0;
    for (let i = 0; i < val; i++) {
      sum += i;
    }
    return sum;
  };
  parentPort.on("message", msg => {
    console.log(`${workerData.name} worker get message: ` + msg);
    parentPort.postMessage(longComputation(msg));
  });
}

// 执行结果
// main thread executing
// Randal worker get message: 10000000000
// main thread get message: 49999999990067860000
// compute time ellapsed: 14.954

线程间如何传输数据

port.postMessag(value[, transferList])

除了 value 之外,postMessage 方法还支持传入 transferList 参数,transferList 是一个 List,支持的数据类型包括 ArrayBufferMessagePort 对象,transferList 中的对象在传输完成后,在发送对象的线程中就不可以继续使用了。

const { Worker, isMainThread, parentPort } = require("worker_threads");
// 主线程
if (isMainThread) {
  const sab = new ArrayBuffer(Int32Array.BYTES_PER_ELEMENT * 100);
  const ia = new Int32Array(sab);

  for (let i = 0; i < ia.length; i++) {
    ia[i] = i;
  }
  console.log("this is the main thread");
  for (let i = 0; i < 1; i++) {
    let w = new Worker(__filename);
    console.log("before transfer: ", sab);
    w.postMessage(null, [sab]);
    setTimeout(() => {
      console.log("after transfer: ", sab);
    }, 1000);
  }
} else {
  console.log("this isn't main thread");
}
// 输出结果
// this is the main thread
// before transfer:  ArrayBuffer { byteLength: 400 }
// this isn't main thread
// after transfer:  ArrayBuffer { byteLength: 0 }

如果 ArrayBuffer 是通过 value 传输的(且在 transferList 中不存在),则传输过去的是副本,如下所示:

w.postMessage(sab);

// 输出结果
// this is the main thread
// before transfer:  ArrayBuffer { byteLength: 400 }
// this isn't main thread
// after transfer:  ArrayBuffer { byteLength: 400 }

线程间如何共享内存

轮到 SharedArrayBuffer 出场了,如果 postMessage 中的 valueSharedArrayBuffer 的话,则线程之间就可以共享内存,如下面例子所示:

const { Worker, isMainThread, parentPort } = require("worker_threads");
// 主线程
if (isMainThread) {
  const sab = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * 5);
  const ia = new Int32Array(sab);

  for (let i = 0; i < ia.length; i++) {
    ia[i] = i;
  }
  for (let i = 0; i < 2; i++) {
    let w = new Worker(__filename);
    w.postMessage(sab);
    w.on("message", () => {
      console.log(ia);
    });
  }
} else {
  parentPort.on("message", msg => {
    const ia = new Int32Array(msg, 0, 1);
    ia[0] = ia[0] + 1;
    parentPort.postMessage("done");
  });
}

// 输出结果
// Int32Array [ 1, 1, 2, 3, 4 ]
// Int32Array [ 2, 1, 2, 3, 4 ]