文章/答案/技术大牛

发布

社区首页 >专栏 >[译]理解 Node.js 的中 Worker Threads

[译]理解 Node.js 的中 Worker Threads

腾讯IVWEB团队

发布于 2020-06-28 02:48:50

2.1K00

代码可运行

文章被收录于专栏：腾讯IVWEB团队的专栏腾讯IVWEB团队的专栏

运行总次数：0

代码可运行

原文：https://nodesource.com/blog/worker-threads-nodejs

理解 Node 的底层对于理解 Workers 是很有必要的。

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

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

一个进程：process 对象是一个全局变量，可在 Node.js 程序中任意地方访问，并提供当前进程的相关信息。

一个线程：单线程意味着在当前进程中同一时刻只有一个指令在执行。

事件循环：这是 Node.js 中需要重点理解的一个部分，尽管 JavaScript 是单线程的，但通过使用回调，promises, async/await 等语法，基于事件循环将对操作系统的操作异步化，使得 Node 拥有异步非阻塞 IO 的特性。

一个 JS 引擎实例：即一个可以运行 JavaScript 代码的程序。

一个 Node.js 实例：即一个可以运行 Node.js 环境的程序。

换言之，Node 运行在单线程上，并且在事件循环中同一时刻只有一个进程的任务被执行，每次同一时刻只会执行一段代码(多段代码不会同时执行)。这是非常有效的，因为这样的机制足够简单，让你在使用 JavaScript 的时候无需担心并发编程的问题。

这样的原因在于 JavaScript 起初是用于客户端的交互(比如 web 页面的交互或表单的验证)，这些逻辑并不需要多线程这样的机制来处理。

所以这也带来了另一个缺点：如果你需要使用 CPU 密集型的任务，比如在内存中使用一个大的数据集进行复杂计算，它会阻塞掉其他进程的任务。同样的，当你在发起一个有 CPU 密集型任务的远程接口请求时，也同样会阻塞掉其他需要被执行的请求。

如果一个函数阻塞了事件循环机制直到这个函数执行完才能执行下一个函数，那么它就被认为是一个阻塞型函数。一个非阻塞的函数是不会阻塞住事件循环进行下一个函数的执行的，它会使用回调通知事件循环函数任务已执行完毕。

最佳实践：不要阻塞事件循环，要让事件循环保持不断运行，并且注意避免使用回阻塞线程的操作比如同步的网络接口调用或死循环。

区分开 CPU 密集型操作与 I/O(input/output) 密集型操作是很重要的。像前面所说的，Node.js 并不会同时执行多段代码，只有 I/O 操作才会同时去执行，因为它们是异步的。

所以 Worker Threads 对于 I/O 密集型操作是没有太大的帮助的，因为异步的 I/O 操作比 worker 更有效率，Wokers 的主要作用是用于提升对于 CPU 密集型操作的性能。

其他方案

此外，目前已经存在很多对于 CPU 密集型操作的解决方案，比如多进程(cluster API)方案，保证了充分利用多核 CPU。

这个方案的好处在于进程之间是相互独立的，如果一个进程出现了问题，并不会影响到其他进程。此外它们还拥有稳定的 API，然而，这也意味着不能同享内存空间，而且进程间通信只能通过 JSON 格式的数据进行交互。

JavaScript 和 Node.js 不会有多线程，理由如下：

所以，人们可能会认为添加一个创建和同步线程的 Node.js 核心模块就可以解决 CPU 密集型操作的需求。

然而并不是，如果添加多线程模块，将会改变语言本身的特性。添加多线程模块作为可用的类或者函数是不可能的。在一些支持多线程的语言比如 Java 中，使用同步特性来使得多个线程之间的同步能够实现。

并且一些数字类型是不够原子性的，这意味着如果你不同步操作它们，在多线程的同时执行计算的情况下，变量的值可能会不断变动，没有确定的值，变量的值可能经过一个线程计算后改变了几个字节，在另一个线程计算后有改变了其他几个字节的数据。比如，在 JavaScript 中一些简单的计算像 0.1 + 0.2 的结果中小数部分有 17 位(小数的最高位数)。

var x = 0.1 + 0.2; // x will be 0.30000000000000004

但是浮点数的计算并不是 100% 精准的。所以如果不同步计算，小数部分的数字就会因为多个线程永远没有一个准确的数字。

最佳实践

所以解决 CPU 密集型操作的性能问题是使用 Worker Threads。浏览器在很久之前就已经有了 Workers 特性了。

单线程下的 Node.js：

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

多线程 Workers 下 Node.js 拥有：

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

就像下图：

Worker_threads 模块允许使用多个线程来同时执行 JavaScript 代码。使用下面这个方式引入：

const worker = require('worker_threads');

Worker Threads 已经被添加到 Node.js 10 版本中，但是仍处于实验阶段。

使用 Worker threads 我们可以在在同一个进程内可以拥有多个 Node.js 实例，并且线程可以不需要跟随父进程的终止的时候才被终止，它可以在任意时刻被终止。当 Worker 线程销毁的时候分配给该 Worker 线程的资源依然没有被释放是一个很不好的操作，这会导致内存泄漏问题，我们也不希望这样。我们希望这些分配资源能够嵌入到 Node.js 中，让 Node.js 有创建线程的能力，并且在线程中创建一个新的 Node.js 实例，本质上就像是在同一个进程中运行多个独立的线程。

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') 的方式监听到。

例子

const { Worker } = require('worker_threads');

const worker = new Worker(`
const { parentPort } = require('worker_threads');
parentPort.once('message',
    message => parentPort.postMessage({ pong: message }));  
`, { eval: true });
worker.on('message', message => console.log(message));      
worker.postMessage('ping');

$ node --experimental-worker test.js
{ pong: ‘ping’ }

上面例子所做的也就是使用 new Worker 创建一个线程，线程中的代码监听了 parentPort 的消息，并且当接收到数据的时候只触发一次回调，将收到的数据传输回父进程中。

你需要使用 --experimental-worker 启动程序因为 Workers 还在实验阶段。

另一个例子：

const {
	Worker, isMainThread, parentPort, workerData
} = require('worker_threads');

if (isMainThread) {
    module.exports = function parseJSAsync(script) {
        return new Promise((resolve, reject) => {
        	const worker = new Worker(filename, {
        		workerData: script
    		});
            worker.on('message', resolve);
            worker.on('error', reject);
            worker.on('exit', (code) => {
                if (code !== 0)
                    reject(new Error(`Worker stopped with exit code ${code}`));
            });
         });
    };
} else {
    const { parse } = require('some-js-parsing-library');
    const script = workerData;
    parentPort.postMessage(parse(script));
}