聊聊Node.js中的多進程和多線程

大家都知道 Node 是單線程的，卻不知它也提供了多進（線）程模塊來加速處理一些特殊任務，本文便帶領大家了解下 Node.js 的多進（線）程，希望對大家有所幫助！

我們都知道 Node.js 采用的是單線程、基于事件驅動的異步 I/O 模型，其特性決定了它無法利用 CPU 多核的優(yōu)勢，也不善于完成一些非 I/O 類型的操作（比如執(zhí)行腳本、AI 計算、圖像處理等），為了解決此類問題，Node.js 提供了常規(guī)的多進（線程）方案（關于進程、線程的討論，可參見筆者的另一篇文章 Node.js 與并發(fā)模型），本文便為大家介紹 Node.js 的多進（線）程機制。

child_process

我們可使用 child_process 模塊創(chuàng)建 Node.js 的子進程，來完成一些特殊的任務（比如執(zhí)行腳本），該模塊主要提供了 exec、execFile、fork、spwan 等方法，下面我們就簡單介紹下這些方法的使用。

exec

const { exec } = require('child_process');  exec('ls -al', (error, stdout, stderr) => {   console.log(stdout); });

該方法根據 options.shell 指定的可執(zhí)行文件處理命令字符串，在命令的執(zhí)行過程中緩存其輸出，直到命令執(zhí)行完成后，再將執(zhí)行結果以回調函數參數的形式返回。

該方法的參數解釋如下：

• command：將要執(zhí)行的命令（比如 ls -al）；

• options：參數設置（可不指定），相關屬性如下：

  • cwd：子進程的當前工作目錄，默認取 process.cwd() 的值；

  • env：環(huán)境變量設置（為鍵值對對象），默認取 process.env 的值；

  • encoding：字符編碼，默認值為：utf8；

  • shell：處理命令字符串的可執(zhí)行文件，Unix 上默認值為 /bin/sh，Windows 上默認值取 process.env.ComSpec 的值（如為空則為 cmd.exe）；比如：

    const { exec } = require('child_process');  exec("print('Hello World!')", { shell: 'python' }, (error, stdout, stderr) => {   console.log(stdout); });

    運行上面的例子將輸出 Hello World!，這等同于子進程執(zhí)行了 python -c "print('Hello World!')" 命令，因此在使用該屬性時需要注意，所指定的可執(zhí)行文件必須支持通過 -c 選項來執(zhí)行相關語句。

    注：碰巧 Node.js 也支持 -c 選項，但它等同于 --check 選項，只用來檢測指定的腳本是否存在語法錯誤，并不會執(zhí)行相關腳本。

  • signal：使用指定的 AbortSignal 終止子進程，該屬性在 v14.17.0 以上可用，比如：

    const { exec } = require('child_process');  const ac = new AbortController(); exec('ls -al', { signal: ac.signal }, (error, stdout, stderr) => {});

    上例中，我們可通過調用 ac.abort() 來提前終止子進程。

  • timeout：子進程的超時時間（如果該屬性的值大于 0，那么當子進程運行時間超過指定值時，將會給子進程發(fā)送屬性 killSignal 指定的終止信號），單位毫米，默認值為 0；

  • maxBuffer：stdout 或 stderr 所允許的最大緩存（二進制），如果超出，子進程將會被殺死，并且將會截斷任何輸出，默認值為 1024 * 1024；

  • killSignal：子進程終止信號，默認值為 SIGTERM；

  • uid：執(zhí)行子進程的 uid；

  • gid：執(zhí)行子進程的 gid；

  • windowsHide：是否隱藏子進程的控制臺窗口，常用于 Windows 系統(tǒng)，默認值為 false；

• callback：回調函數，包含 error、stdout、stderr 三個參數：

  • error：如果命令行執(zhí)行成功，值為 null，否則值為 Error 的一個實例，其中 error.code 為子進程的退出的錯誤碼，error.signal 為子進程終止的信號；
  • stdout 和 stderr：子進程的 stdout 和 stderr，按照 encoding 屬性的值進行編碼，如果 encoding 的值為 buffer，或者 stdout、stderr 的值是一個無法識別的字符串，將按照 buffer 進行編碼。

execFile

const { execFile } = require('child_process');  execFile('ls', ['-al'], (error, stdout, stderr) => {   console.log(stdout); });

該方法的功能類似于 exec，唯一的區(qū)別是 execFile 在默認情況下直接用指定的可執(zhí)行文件（即參數 file 的值）處理命令，這使得其效率略高于 exec（如果查看 shell 的處理邏輯，筆者感覺這效率可忽略不計）。

該方法的參數解釋如下：

• file：可執(zhí)行文件的名字或路徑；

• args：可執(zhí)行文件的參數列表；

• options：參數設置（可不指定），相關屬性如下：

  • shell：值為 false 時表示直接用指定的可執(zhí)行文件（即參數 file 的值）處理命令，值為 true 或其它字符串時，作用等同于 exec 中的 shell，默認值為 false；
  • windowsVerbatimArguments：在 Windows 中是否對參數進行引號或轉義處理，在 Unix 中將忽略該屬性，默認值為 false；
  • 屬性 cwd、env、encoding、timeout、maxBuffer、killSignal、uid、gid、windowsHide、signal 在上文中已介紹，此處不再重述。

• callback：回調函數，等同于 exec 中的 callback，此處不再闡述。

fork

const { fork } = require('child_process');  const echo = fork('./echo.js', {   silent: true }); echo.stdout.on('data', (data) => {   console.log(`stdout: ${data}`); });  echo.stderr.on('data', (data) => {   console.error(`stderr: ${data}`); });  echo.on('close', (code) => {   console.log(`child process exited with code ${code}`); });

該方法用于創(chuàng)建新的 Node.js 實例以執(zhí)行指定的 Node.js 腳本，與父進程之間以 IPC 方式進行通信。

該方法的參數解釋如下：

• modulePath：要運行的 Node.js 腳本路徑；

• args：傳遞給 Node.js 腳本的參數列表；

• options：參數設置（可不指定），相關屬性如：

  • detached：參見下文對 spwan 中 options.detached 的說明；

  • execPath：創(chuàng)建子進程的可執(zhí)行文件；

  • execArgv：傳遞給可執(zhí)行文件的字符串參數列表，默認取 process.execArgv 的值；

  • serialization：進程間消息的序列號類型，可用值為 json 和 advanced，默認值為 json；

  • slient： 如果為 true，子進程的 stdin、stdout 和 stderr 將通過管道傳遞給父進程，否則將繼承父進程的 stdin、stdout 和 stderr；默認值為 false；

  • stdio：參見下文對 spwan 中 options.stdio 的說明。這里需要注意的是：

    • 如果指定了該屬性，將忽略 slient 的值；
    • 必須包含一個值為 ipc 的選項（比如 [0, 1, 2, 'ipc']），否則將拋出異常。

  • 屬性 cwd、env、uid、gid、windowsVerbatimArguments、signal、timeout、killSignal 在上文中已介紹，此處不再重述。

spwan

const { spawn } = require('child_process');  const ls = spawn('ls', ['-al']); ls.stdout.on('data', (data) => {   console.log(`stdout: ${data}`); });  ls.stderr.on('data', (data) => {   console.error(`stderr: ${data}`); });  ls.on('close', (code) => {   console.log(`child process exited with code ${code}`); });

該方法為 child_process 模塊的基礎方法，exec、execFile、fork 最終都會調用 spawn 來創(chuàng)建子進程。

該方法的參數解釋如下：

• command：可執(zhí)行文件的名字或路徑；

• args：傳遞給可執(zhí)行文件的參數列表；

• options：參數設置（可不指定），相關屬性如下：

  • argv0：發(fā)送給子進程 argv[0] 的值，默認取參數 command 的值；

  • detached：是否允許子進程可以獨立于父進程運行（即父進程退出后，子進程可以繼續(xù)運行），默認值為 false，其值為 true 時，各平臺的效果如下所述：

    • 在 Windows 系統(tǒng)中，父進程退出后，子進程可以繼續(xù)運行，并且子進程擁有自己的控制臺窗口（該特性一旦啟動后，在運行過程中將無法更改）；
    • 在非 Windows 系統(tǒng)中，子進程將作為新進程會話組的組長，此刻不管子進程是否與父進程分離，子進程都可以在父進程退出后繼續(xù)運行。

    需要注意的是，如果子進程需要執(zhí)行長時間的任務，并且想要父進程提前退出，需要同時滿足以下幾點：

    • 調用子進程的 unref 方法從而將子進程從父進程的事件循環(huán)中剔除；
    • detached 設置為 true；
    • stdio 為 ignore。

    比如下面的例子：

    // hello.js const fs = require('fs'); let index = 0; function run() {   setTimeout(() => {     fs.writeFileSync('./hello', `index: ${index}`);     if (index < 10) {       index += 1;       run();     }   }, 1000); } run();  // main.js const { spawn } = require('child_process'); const child = spawn('node', ['./hello.js'], {   detached: true,   stdio: 'ignore' }); child.unref();

  • stdio：子進程標準輸入輸出配置，默認值為 pipe，值為字符串或數組：

    • 值為字符串時，會將其轉換為含有三個項的數組（比如 pipe 被轉換為 ['pipe', 'pipe', 'pipe']），可用值為 pipe、overlapped、ignore、inherit；
    • 值為數組時，其中數組的前三項分別代表對 stdin、stdout 和 stderr 的配置，每一項的可用值為 pipe、overlapped、ignore、inherit、ipc、Stream 對象、正整數（在父進程打開的文件描述符）、null（如位于數組的前三項，等同于 pipe，否則等同于 ignore）、undefined（如位于數組的前三項，等同于 pipe，否則等同于 ignore）。

  • 屬性 cwd、env、uid、gid、serialization、shell（值為 boolean 或 string）、windowsVerbatimArguments、windowsHide、signal、timeout、killSignal 在上文中已介紹，此處不再重述。

小結

上文對 child_process 模塊中主要方法的使用進行了簡短介紹，由于 execSync、execFileSync、forkSync、spwanSync 方法是 exec、execFile、spwan 的同步版本，其參數并無任何差異，故不再重述。

cluster

通過 cluster 模塊我們可以創(chuàng)建 Node.js 進程集群，通過 Node.js 進程進群，我們可以更加充分地利用多核的優(yōu)勢，將程序任務分發(fā)到不同的進程中以提高程序的執(zhí)行效率；下面將通過例子為大家介紹 cluster 模塊的使用：

const http = require('http'); const cluster = require('cluster'); const numCPUs = require('os').cpus().length;  if (cluster.isPrimary) {   for (let i = 0; i < numCPUs; i++) {     cluster.fork();   } } else {   http.createServer((req, res) => {     res.writeHead(200);     res.end(`${process.pid}n`);   }).listen(8000); }

上例通過 cluster.isPrimary 屬性判斷（即判斷當前進程是否為主進程）將其分為兩個部分：

• 為真時，根據 CPU 內核的數量并通過 cluster.fork 調用來創(chuàng)建相應數量的子進程；
• 為假時，創(chuàng)建一個 HTTP server，并且每個 HTTP server 都監(jiān)聽同一個端口（此處為 8000）。

運行上面的例子，并在瀏覽器中訪問 http://localhost:8000/，我們會發(fā)現每次訪問返回的 pid 都不一樣，這說明了請求確實被分發(fā)到了各個子進程。Node.js 默認采用的負載均衡策略是輪詢調度，可通過環(huán)境變量 NODE_CLUSTER_SCHED_POLICY 或 cluster.schedulingPolicy 屬性來修改其負載均衡策略：

NODE_CLUSTER_SCHED_POLICY = rr // 或 none  cluster.schedulingPolicy = cluster.SCHED_RR; // 或 cluster.SCHED_NONE

另外需要注意的是，雖然每個子進程都創(chuàng)建了 HTTP server，并都監(jiān)聽了同一個端口，但并不代表由這些子進程自由競爭用戶請求，因為這樣無法保證所有子進程的負載達到均衡。所以正確的流程應該是由主進程監(jiān)聽端口，然后將用戶請求根據分發(fā)策略轉發(fā)到具體的子進程進行處理。

由于進程之間是相互隔離的，因此進程之間一般通過共享內存、消息傳遞、管道等機制進行通訊。Node.js 則是通過消息傳遞來完成父子進程之間的通信，比如下面的例子：

const http = require('http'); const cluster = require('cluster'); const numCPUs = require('os').cpus().length;  if (cluster.isPrimary) {   for (let i = 0; i < numCPUs; i++) {     const worker = cluster.fork();     worker.on('message', (message) => {       console.log(`I am primary(${process.pid}), I got message from worker: "${message}"`);       worker.send(`Send message to worker`)     });   } } else {   process.on('message', (message) => {     console.log(`I am worker(${process.pid}), I got message from primary: "${message}"`)   });   http.createServer((req, res) => {     res.writeHead(200);     res.end(`${process.pid}n`);     process.send('Send message to primary');   }).listen(8000); }

運行上面的例子，并訪問 http://localhost:8000/，再查看終端，我們會看到類似下面的輸出：

I am primary(44460), I got message from worker: "Send message to primary" I am worker(44461), I got message from primary: "Send message to worker" I am primary(44460), I got message from worker: "Send message to primary" I am worker(44462), I got message from primary: "Send message to worker"

利用該機制，我們可以監(jiān)聽各子進程的狀態(tài)，以便在某個子進程出現意外后，能夠及時對其進行干預，以保證服務的可用性。

cluster 模塊的接口非常簡單，為了節(jié)省篇幅，這里只對 cluster.setupPrimary 方法做一些特別聲明，其它方法請查看官方文檔：

• cluster.setupPrimary 調用后，相關設置將同步到在 cluster.settings 屬性中，并且每次調用都基于當前 cluster.settings 屬性的值；
• cluster.setupPrimary 調用后，對已運行的子進程沒有影響，只影響后續(xù)的 cluster.fork 調用；
• cluster.setupPrimary 調用后，不影響后續(xù)傳遞給 cluster.fork 調用的 env 參數；
• cluster.setupPrimary 只能在主進程中使用。

worker_threads

前文我們對 cluster 模塊進行了介紹，通過它我們可以創(chuàng)建 Node.js 進程集群以提高程序的運行效率，但 cluster 基于多進程模型，進程間高成本的切換以及進程間資源的隔離，會隨著子進程數量的增加，很容易導致因系統(tǒng)資源緊張而無法響應的問題。為解決此類問題，Node.js 提供了 worker_threads，下面我們通過具體的例子對該模塊的使用進行簡單介紹：

// server.js const http = require('http'); const { Worker } = require('worker_threads');  http.createServer((req, res) => {   const httpWorker = new Worker('./http_worker.js');   httpWorker.on('message', (result) => {     res.writeHead(200);     res.end(`${result}n`);   });   httpWorker.postMessage('Tom'); }).listen(8000);  // http_worker.js const { parentPort } = require('worker_threads');  parentPort.on('message', (name) => {   parentPort.postMessage(`Welcone ${name}!`); });

上例展示了 worker_threads 的簡單使用，在使用 worker_threads 的過程中，需要注意以下幾點：

• 通過 worker_threads.Worker 創(chuàng)建 Worker 實例，其中 Worker 腳本既可以為一個獨立的 JavaScript 文件，也可以為字符串，比如上例可修改為：

  const code = "const { parentPort } = require('worker_threads'); parentPort.on('message', (name) => {parentPort.postMessage(`Welcone ${name}!`);})"; const httpWorker = new Worker(code, { eval: true });

• 通過 worker_threads.Worker 創(chuàng)建 Worker 實例時，可以通過指定 workerData 的值來設置 Worker 子線程的初始元數據，比如：

  // server.js const { Worker } = require('worker_threads'); const httpWorker = new Worker('./http_worker.js', { workerData: { name: 'Tom'} });  // http_worker.js const { workerData } = require('worker_threads'); console.log(workerData);

• 通過 worker_threads.Worker 創(chuàng)建 Worker 實例時，可通過設置 SHARE_ENV 以實現在 Worker 子線程與主線程之間共享環(huán)境變量的需求，比如：

  const { Worker, SHARE_ENV } = require('worker_threads'); const worker = new Worker('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV }); worker.on('exit', () => {   console.log(process.env.SET_IN_WORKER); });

• 不同于 cluster 中進程間的通信機制，worker_threads 采用的 MessageChannel 來進行線程間的通信：

  • Worker 子線程通過 parentPort.postMessage 方法發(fā)送消息給主線程，并通過監(jiān)聽 parentPort 的 message 事件來處理來自主線程的消息；
  • 主線程通過 Worker 子線程實例（此處為 httpWorker，以下均以此代替 Worker 子線程）的 postMessage 方法發(fā)送消息給 httpWorker，并通過監(jiān)聽 httpWorker 的 message 事件來處理來自 Worker 子線程的消息。

在 Node.js 中，無論是 cluster 創(chuàng)建的子進程，還是 worker_threads 創(chuàng)建的 Worker 子線程，它們都擁有屬于自己的 V8 實例以及事件循環(huán)，所不同的是：

• 子進程之間的內存空間是互相隔離的，而 Worker 子線程共享所屬進程的內存空間；
• 子進程之間的切換成本要遠遠高于 Worker 子線程之間的切換成本。

盡管看起來 Worker 子線程比子進程更高效，但 Worker 子線程也有不足的地方，即cluster 提供了負載均衡，而 worker_threads 則需要我們自行完成負載均衡的設計與實現。

總結

本文介紹了 Node.js 中 child_process、cluster 和 worker_threads 三個模塊的使用，通過這三個模塊，我們可以充分利用 CPU 多核的優(yōu)勢，并以多進（線）程的模式來高效地解決一些特殊任務（比如 AI、圖片處理等）的運行效率。每個模塊都有其適用的場景，文中僅對其基本使用進行了說明，如何結合自己的問題進行高效地運用，還需要大家自行摸索。最后，本文若有紕漏之處，還望大家能夠指正，祝大家快樂編碼每一天。