快捷導(dǎo)航

Nodejs 構(gòu)建Cluster集群多線程Worker threads

更新時(shí)間：2022年10月21日 11:19:35 作者：SaraiNoQ

這篇文章主要為大家介紹了Nodejs 構(gòu)建Cluster集群多線程Worker threads示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

前兩天我們介紹了使用 Nodejs 中的 child_process 模塊創(chuàng)建多個(gè)子進(jìn)程，同時(shí)利用進(jìn)程間通信的API構(gòu)建了一個(gè)集群式的Web服務(wù)器。實(shí)際上，你可以通過 cluster 模塊更方便的完成這一操作。

但是，cluster 創(chuàng)建的進(jìn)程之間無法共享內(nèi)存，通信必須使用 JSON 格式，有一定的局限性和性能問題。如果你不想要進(jìn)程隔離，可以使用 worker_thread 模塊，它允許在一個(gè) Node.js 實(shí)例中運(yùn)行多個(gè)應(yīng)用程序線程。相比創(chuàng)建多個(gè)進(jìn)程更輕量，并且可以共享內(nèi)存。

進(jìn)程間通過傳輸 ArrayBuffer 實(shí)例或共享 SharedArrayBuffer 實(shí)例來做到這一點(diǎn)，對(duì)數(shù)據(jù)格式?jīng)]有太多要求。但是要注意，數(shù)據(jù)中不能包含函數(shù)。

Cluster 多進(jìn)程

我們可以使用 cluster 模塊提供的API重構(gòu)昨天的案例：

// master.js
const cl = require("cluster");
const cpus = require("os").cpus().length;
// 修改默認(rèn)的 fork() 方法配置
cl.setupPrimary({
  exec: 'worker.js'
});
for(let i = 0; i < cpus; i++) {
  cl.fork();
};
cl.on('listening', (data) => {
  console.log(`listenning on: ${data.id}--${data.process.pid}`);
});
cl.on('exit', (data, code, signal) => {
  console.log(`exited: ${data.id}--${data.process.pid}, kill code: $[code], signal: ${signal}`);
  cl.fork();
});

子進(jìn)程依舊使用昨天的代碼：

const http = require("http");
const server = http.createServer((req, res) => {
  res.writeHead(200, {
    "Content-Type": "text/plain"
  });
  res.end("Hello,World!" + process.pid);
  // 拋出異常，捕獲后終止進(jìn)程
  throw new Error('throw exception');
}).listen(1337);
// 捕獲異常后終止進(jìn)程
process.on('uncaughtException', (err) => {
  // 停止接收新的連接
  server.close((data) => {
    console.log(`worker: ${process.pid} is stopping!`);
    process.exit(1);
  })
  // 避免長連接請(qǐng)求長時(shí)間無法終止，5s后自動(dòng)終止
  setTimeout(() => {
    process.exit(1);
  }, 5000)
});

執(zhí)行 node master.js，會(huì)得到與昨天利用 child_process 模塊創(chuàng)建子進(jìn)程集群相同的效果。

同樣，你可以使用官方推薦的寫法，利用 cluster.isPrimary 和 cluster.isWorker 來判斷當(dāng)前進(jìn)程是否為主進(jìn)程：

const cluster = require('node:cluster');
const http = require('node:http');
const numCPUs = require('node:os').cpus().length;
const process = require('node:process');
if (cluster.isPrimary) {
  console.log(`Primary ${process.pid} is running`);
  // Fork workers.
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }
  cluster.on('exit', (worker, code, signal) => {
    console.log(`worker ${worker.process.pid} died`);
  });
} else {
  // Workers can share any TCP connection
  // In this case it is an HTTP server
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');
  }).listen(1337);
  console.log(`Worker ${process.pid} started`);
};

實(shí)現(xiàn)原理

事實(shí)上，cluster 模塊就是將 child_process 和 net 模塊的API組合起來實(shí)現(xiàn)的。cluster啟動(dòng)時(shí)，進(jìn)程會(huì)在內(nèi)部啟動(dòng)TCP服務(wù)器。而在調(diào)用 cluster.fork() 復(fù)制子進(jìn)程時(shí)，會(huì)將這個(gè)TCP服務(wù)器端 Socket 的句柄發(fā)送給工作進(jìn)程。如果進(jìn)程是通過 cluster.fork() 復(fù)制出來的，那么它的環(huán)境變量里就存在 NODE_UNIQUE_ID。如果工作進(jìn)程中存在 listen() 偵聽網(wǎng)絡(luò)端口的調(diào)用，它將拿到該句柄，再通過 SO_REUSEADDR 端口重用，從而實(shí)現(xiàn)多個(gè)子進(jìn)程共享端口。對(duì)于正常方式啟動(dòng)的進(jìn)程，則不存在句柄共享和傳遞等過程。

在 cluster 內(nèi)部隱式創(chuàng)建TCP服務(wù)器的方式對(duì)使用者是透明的，你不需要自己手動(dòng)去實(shí)現(xiàn)句柄的傳遞，但也正是因此，它無法像使用 child_process 那樣靈活。在 child_process 中你可以自行控制句柄的傳送，因此可以靈活地控制工作進(jìn)程，甚至控制多組工作進(jìn)程。

cluster事件

Event: disconnect 主進(jìn)程和工作進(jìn)程之間IPC通道斷開后會(huì)觸發(fā)該事件。
Event: exit 有工作進(jìn)程退出時(shí)觸發(fā)該事件。
Event: fork 復(fù)制一個(gè)工作進(jìn)程后觸發(fā)該事件。
Event: listening 工作進(jìn)程中調(diào)用 listen() 后，發(fā)送該消息給主進(jìn)程，主進(jìn)程收到后，觸發(fā)該事件。
Event: message
Event: online fork好一個(gè)工作進(jìn)程后，工作進(jìn)程主動(dòng)發(fā)送該消息給主進(jìn)程，主進(jìn)程收到消息后，觸發(fā)該事件。
Event: setup .setupPrimary() 方法執(zhí)行后觸發(fā)

?? 這些事件大多跟 child_process 模塊的事件相關(guān)，在進(jìn)程間消息傳遞的基礎(chǔ)上完成的封裝。

使用 Node 構(gòu)建集群能夠充分利用多核CPU的計(jì)算性能，而 child_process 模塊的進(jìn)程間通信和多種事件能夠極大提升Node的穩(wěn)定性。但進(jìn)程間無法共享資源，進(jìn)程間通信有局限性和性能問題。此時(shí)就需要引入更輕量級(jí)的線程了。

Worker threads多線程

V8 多線程模型

眾所周知，JavaScript 在運(yùn)行時(shí)是單線程的。但 JavaScript 的 Runtime V8 引擎卻不是單線程的。大致包括以下幾個(gè)線程：

JavaScript 主線程：編譯、執(zhí)行代碼。
編譯線程：當(dāng)主線程在執(zhí)行時(shí)，編譯線程可以優(yōu)化代碼。
Profiler 線程：記錄方法耗時(shí)的線程。
其它線程：比如支持并行 GC 的多線程。
libuv線程池，默認(rèn)四個(gè)線程，全局共享，可以將異步操作和計(jì)算密集任務(wù)交給它執(zhí)行。

對(duì)于 Node 來說，crypto 這種 CPU 密集和 fs 這種 I/O 密集的任務(wù)是在 libuv線程池 中進(jìn)行的。其執(zhí)行模型是單獨(dú)創(chuàng)建一個(gè)進(jìn)程，在這個(gè)進(jìn)程中同步執(zhí)行任務(wù)，然后將結(jié)果返回到 Event Loop 中，Event Loop 可以通過回調(diào)函數(shù)獲取并使用結(jié)果。

const fs = require("fs");
fs.writeFile('./target.txt', 'hello Node.js', (err) => {
  if (err) throw err;
  console.log('文件已被保存');
});

使用非阻塞方法，長耗時(shí)的方法不會(huì)阻塞主進(jìn)程之后的代碼，只需告訴 Worker Pool 去執(zhí)行該命令，并將結(jié)果返回給預(yù)先設(shè)置好的回調(diào)函數(shù)，在計(jì)算完成時(shí)觸發(fā)即可。

?? 由于 Worker Pool 運(yùn)行在 libuv線程池 中，主線程的 Event Loop 不會(huì)被阻塞。能夠充分利用 CPU 資源。

多線程支持

Node v10.5.0 提供了 Worker threads 模塊，開始支持多線程編程。在創(chuàng)建出的每個(gè)工作線程中，都會(huì)包含 V8 和 libuv，即都包含Event Loop：

你可以通過下面這段簡單的代碼來體驗(yàn)一下：

// main.js
const { Worker, isMainThread } = require('worker_threads');
if (isMainThread) {
  console.log("I'm main thread: ", isMainThread);
  // create subThread
  new Worker(__filename);
}
else {
  console.log("I'm not main thread: ", isMainThread);
  // subThread destroy
}

我們在主線程中調(diào)用new方法創(chuàng)建了一個(gè)子線程，子線程執(zhí)行完自動(dòng)銷毀。最后執(zhí)行結(jié)果如下：