Node.js 多線程完全指南總結

很多人都想知道單線程的 Node.js 怎么能與多線程后端競爭。考慮到其所謂的單線程特性，許多大公司選擇 Node 作為其后端似乎違反直覺。要想知道原因，必須理解其單線程的真正含義。

JavaScript 的設計非常適合在網上做比較簡單的事情，比如驗證表單，或者說創建彩虹色的鼠標軌跡。 在2009年，Node.js的創始人 Ryan Dahl使開發人員可以用該語言編寫后端代碼。

通常支持多線程的后端語言具有各種機制，用于在線程和其他面向線程的功能之間同步數據。要向 JavaScript 添加對此類功能的支持，需要修改整個語言，這不是 Dahl 的目標。為了讓純 JavaScript 支持多線程，他必須想一個變通方法。接下來讓我們探索一下其中的奧秘……

Node.js 是如何工作的

Node.js 使用兩種線程：event loop 處理的主線程和 worker pool 中的幾個輔助線程。

事件循環是一種機制，它采用回調（函數）并注冊它們，準備在將來的某個時刻執行。它與相關的 JavaScript 代碼在同一個線程中運行。當 JavaScript 操作阻塞線程時，事件循環也會被阻止。

工作池是一種執行模型，它產生并處理單獨的線程，然后同步執行任務，并將結果返回到事件循環。事件循環使用返回的結果執行提供的回調。

簡而言之，它負責異步 I/O操作 —— 主要是與系統磁盤和網絡的交互。它主要由諸如 fs（I/O 密集）或 crypto（CPU 密集）等模塊使用。工作池用 libuv 實現，當 Node 需要在 JavaScript 和 C++ 之間進行內部通信時，會導致輕微的延遲，但這幾乎不可察覺。

基于這兩種機制，我們可以編寫如下代碼：

fs.readFile(path.join(__dirname, './package.json'), (err, content) => {
 if (err) {
  return null;
 }

 console.log(content.toString());
});

前面提到的 fs 模塊告訴工作池使用其中一個線程來讀取文件的內容，并在完成后通知事件循環。然后事件循環獲取提供的回調函數，并用文件的內容執行它。

以上是非阻塞代碼的示例，我們不必同步等待某事的發生。只需告訴工作池去讀取文件，并用結果去調用提供的函數即可。由于工作池有自己的線程，因此事件循環可以在讀取文件時繼續正常執行。

在不需要同步執行某些復雜操作時，這一切都相安無事：任何運行時間太長的函數都會阻塞線程。如果應用程序中有大量這類功能，就可能會明顯降低服務器的吞吐量，甚至完全凍結它。在這種情況下，無法繼續將工作委派給工作池。

在需要對數據進行復雜的計算時（如AI、機器學習或大數據）無法真正有效地使用 Node.js，因為操作阻塞了主（且唯一）線程，使服務器無響應。在 Node.js v10.5.0 發布之前就是這種情況，在這一版本增加了對多線程的支持。

簡介：worker_threads

worker_threads 模塊允許我們創建功能齊全的多線程 Node.js 程序。

thread worker 是在單獨的線程中生成的一段代碼（通常從文件中取出）。

注意，術語 thread worker，worker 和 thread 經常互換使用，他們都指的是同一件事。

要想使用 thread worker，必須導入 worker_threads 模塊。讓我們先寫一個函數來幫助我們生成這些thread worker，然后再討論它們的屬性。

type WorkerCallback = (err: any, result?: any) => any;

export function runWorker(path: string, cb: WorkerCallback, workerData: object | null = null) {
 const worker = new Worker(path, { workerData });

 worker.on('message', cb.bind(null, null));
 worker.on('error', cb);

 worker.on('exit', (exitCode) => {
  if (exitCode === 0) {
   return null;
  }

  return cb(new Error(`Worker has stopped with code ${exitCode}`));
 });

 return worker;
}

要創建一個 worker，首先必須創建一個 Worker 類的實例。它的第一個參數提供了包含 worker 的代碼的文件的路徑；第二個參數提供了一個名為 workerData 的包含一個屬性的對象。這是我們希望線程在開始運行時可以訪問的數據。

請注意：不管你是用的是 JavaScript， 還是最終要轉換為 JavaScript 的語言（例如，TypeScript），路徑應該始終引用帶有 .js 或 .mjs 擴展名的文件。

我還想指出為什么使用回調方法，而不是返回在觸發 message 事件時將解決的 promise。這是因為 worker 可以發送許多 message 事件，而不是一個。

正如你在上面的例子中所看到的，線程間的通信是基于事件的，這意味著我們設置了 worker 在發送給定事件后調用的偵聽器。

以下是最常見的事件：

worker.on('error', (error) => {});

只要 worker 中有未捕獲的異常，就會發出 error 事件。然后終止 worker，錯誤可以作為提供的回調中的第一個參數。

worker.on('exit', (exitCode) => {});

在 worker 退出時會發出 exit 事件。如果在worker中調用了 process.exit()，那么 exitCode 將被提供給回調。如果 worker 以 worker.terminate() 終止，則代碼為1。

worker.on('online', () => {});

只要 worker 停止解析 JavaScript 代碼并開始執行，就會發出 online 事件。它不常用，但在特定情況下可以提供信息。

worker.on('message', (data) => {});

只要 worker 將數據發送到父線程，就會發出 message 事件。

現在讓我們來看看如何在線程之間共享數據。

在線程之間交換數據

要將數據發送到另一個線程，可以用 port.postMessage() 方法。它的原型如下：

port.postMessage(data[, transferList])

port 對象可以是 parentPort，也可以是 MessagePort 的實例 —— 稍后會詳細講解。

數據參數

第一個參數 —— 這里被稱為 data —— 是一個被復制到另一個線程的對象。它可以是復制算法所支持的任何內容。

數據由結構化克隆算法進行復制。引用自 Mozilla：

它通過遞歸輸入對象來進行克隆，同時保持之前訪問過的引用的映射，以避免無限遍歷循環。

該算法不復制函數、錯誤、屬性描述符或原型鏈。還需要注意的是，以這種方式復制對象與使用 JSON 不同，因為它可以包含循環引用和類型化數組，而 JSON 不能。

由于能夠復制類型化數組，該算法可以在線程之間共享內存。

在線程之間共享內存

人們可能會說像 cluster 或 child_process 這樣的模塊在很久以前就開始使用線程了。這話對，也不對。

cluster 模塊可以創建多個節點實例，其中一個主進程在它們之間對請求進行路由。集群能夠有效地增加服務器的吞吐量；但是我們不能用 cluster 模塊生成一個單獨的線程。

人們傾向于用 PM2 這樣的工具來集中管理他們的程序，而不是在自己的代碼中手動執行，如果你有興趣，可以研究一下如何使用 cluster 模塊。

child_process 模塊可以生成任何可執行文件，無論它是否是用 JavaScript 寫的。它和 worker_threads 非常相似，但缺少后者的幾個重要功能。

具體來說 thread workers 更輕量，并且與其父線程共享相同的進程 ID。它們還可以與父線程共享內存，這樣可以避免對大的數據負載進行序列化，從而更有效地來回傳遞數據。

現在讓我們看一下如何在線程之間共享內存。為了共享內存，必須將 ArrayBuffer 或 SharedArrayBuffer 的實例作為數據參數發送到另一個線程。

這是一個與其父線程共享內存的 worker：

import { parentPort } from 'worker_threads';

parentPort.on('message', () => {
 const numberOfElements = 100;
 const sharedBuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * numberOfElements);
 const arr = new Int32Array(sharedBuffer);

 for (let i = 0; i < numberOfElements; i += 1) {
  arr[i] = Math.round(Math.random() * 30);
 }

 parentPort.postMessage({ arr });
});

首先，我們創建一個 SharedArrayBuffer，其內存需要包含100個32位整數。接下來創建一個 Int32Array 實例，它將用緩沖區來保存其結構，然后用一些隨機數填充數組并將其發送到父線程。

在父線程中：

import path from 'path';

import { runWorker } from '../run-worker';

const worker = runWorker(path.join(__dirname, 'worker.js'), (err, { arr }) => {
 if (err) {
  return null;
 }

 arr[0] = 5;
});

worker.postMessage({});

把 arr [0] 的值改為5，實際上會在兩個線程中修改它。

當然，通過共享內存，我們冒險在一個線程中修改一個值，同時也在另一個線程中進行了修改。但是我們在這個過程中也得到了一個好處：該值不需要進行序列化就可以另一個線程中使用，這極大地提高了效率。只需記住管理數據正確的引用，以便在完成數據處理后對其進行垃圾回收。

共享一個整數數組固然很好，但我們真正感興趣的是共享對象 —— 這是存儲信息的默認方式。不幸的是，沒有 SharedObjectBuffer 或類似的東西，但我們可以自己創建一個類似的結構。

transferList參數

transferList 中只能包含 ArrayBuffer 和 MessagePort。一旦它們被傳送到另一個線程，就不能再次被傳送了；因為內存里的內容已經被移動到了另一個線程。

目前，還不能通過 transferList（可以使用 child_process 模塊）來傳輸網絡套接字。

創建通信渠道

線程之間的通信是通過 port 進行的，port 是 MessagePort 類的實例，并啟用基于事件的通信。

使用 port 在線程之間進行通信的方法有兩種。第一個是默認值，這個方法比較容易。在 worker 的代碼中，我們從worker_threads 模塊導入一個名為 parentPort 的對象，并使用對象的 .postMessage() 方法將消息發送到父線程。

這是一個例子：

import { parentPort } from 'worker_threads';
const data = {
 // ...
};

parentPort.postMessage(data);

parentPort 是 Node.js 在幕后創建的 MessagePort 實例，用于與父線程進行通信。這樣就可以用 parentPort 和 worker 對象在線程之間進行通信。

線程間的第二種通信方式是創建一個 MessageChannel 并將其發送給 worker。以下代碼是如何創建一個新的 MessagePort 并與我們的 worker 共享它：

import path from 'path';
import { Worker, MessageChannel } from 'worker_threads';

const worker = new Worker(path.join(__dirname, 'worker.js'));

const { port1, port2 } = new MessageChannel();

port1.on('message', (message) => {
 console.log('message from worker:', message);
});

worker.postMessage({ port: port2 }, [port2]);

在創建 port1 和 port2 之后，我們在 port1 上設置事件監聽器并將 port2 發送給 worker。我們必須將它包含在 transferList 中，以便將其傳輸給 worker 。

在 worker 內部：

import { parentPort, MessagePort } from 'worker_threads';

parentPort.on('message', (data) => {
 const { port }: { port: MessagePort } = data;

 port.postMessage('heres your message!');
});

這樣，我們就能使用父線程發送的 port 了。

使用 parentPort 不一定是錯誤的方法，但最好用 MessageChannel 的實例創建一個新的 MessagePort，然后與生成的 worker 共享它。

請注意，在后面的例子中，為了簡便起見，我用了 parentPort。

使用 worker 的兩種方式

可以通過兩種方式使用 worker。第一種是生成一個 worker，然后執行它的代碼，并將結果發送到父線程。通過這種方法，每當出現新任務時，都必須重新創建一個工作者。

第二種方法是生成一個 worker 并為 message 事件設置監聽器。每次觸發 message 時，它都會完成工作并將結果發送回父線程，這會使 worker 保持活動狀態以供以后使用。

Node.js 文檔推薦第二種方法，因為在創建 thread worker 時需要創建虛擬機并解析和執行代碼，這會產生比較大的開銷。所以這種方法比不斷產生新 worker 的效率更高。

這種方法被稱為工作池，因為我們創建了一個工作池并讓它們等待，在需要時調度 message 事件來完成工作。

以下是一個產生、執行然后關閉 worker 例子：

import { parentPort } from 'worker_threads';

const collection = [];

for (let i = 0; i < 10; i += 1) {
 collection[i] = i;
}

parentPort.postMessage(collection);

將 collection 發送到父線程后，它就會退出。

下面是一個 worker 的例子，它可以在給定任務之前等待很長一段時間：

import { parentPort } from 'worker_threads';

parentPort.on('message', (data: any) => {
 const result = doSomething(data);

 parentPort.postMessage(result);
});

worker_threads 模塊中可用的重要屬性

worker_threads 模塊中有一些可用的屬性：

isMainThread

當不在工作線程內操作時，該屬性為 true 。如果你覺得有必要，可以在 worker 文件的開頭包含一個簡單的 if 語句，以確保它只作為 worker 運行。

import { isMainThread } from 'worker_threads';

if (isMainThread) {
 throw new Error('Its not a worker');
}

workerData

產生線程時包含在 worker 的構造函數中的數據。

const worker = new Worker(path, { workerData });

在工作線程中：

import { workerData } from 'worker_threads';

console.log(workerData.property);

parentPort

前面提到的 MessagePort 實例，用于與父線程通信。

threadId

分配給 worker 的唯一標識符。

現在我們知道了技術細節，接下來實現一些東西并在實踐中檢驗學到的知識。

實現 setTimeout

setTimeout 是一個無限循環，顧名思義，用來檢測程序運行時間是否超時。它在循環中檢查起始時間與給定毫秒數之和是否小于實際日期。

import { parentPort, workerData } from 'worker_threads';

const time = Date.now();

while (true) {
  if (time + workerData.time <= Date.now()) {
    parentPort.postMessage({});
    break;
  }
}

這個特定的實現產生一個線程，然后執行它的代碼，最后在完成后退出。

接下來實現使用這個 worker 的代碼。首先創建一個狀態，用它來跟蹤生成的 worker：

const timeoutState: { [key: string]: Worker } = {};

然后時負責創建 worker 并將其保存到狀態的函數：

export function setTimeout(callback: (err: any) => any, time: number) {
 const id = uuidv4();

 const worker = runWorker(
  path.join(__dirname, './timeout-worker.js'),
  (err) => {
   if (!timeoutState[id]) {
    return null;
   }

   timeoutState[id] = null;

   if (err) {
    return callback(err);
   }

   callback(null);
  },
  {
   time,
  },
 );

 timeoutState[id] = worker;

 return id;
}

首先，我們使用 UUID 包為 worker 創建一個唯一的標識符，然后用先前定義的函數 runWorker 來獲取 worker。我們還向 worker 傳入一個回調函數，一旦 worker 發送了數據就會被觸發。最后，把 worker 保存在狀態中并返回 id。

在回調函數中，我們必須檢查該 worker 是否仍然存在于該狀態中，因為有可能會 cancelTimeout()，這將會把它刪除。如果確實存在，就把它從狀態中刪除，并調用傳給 setTimeout 函數的 callback。

cancelTimeout 函數使用 .terminate() 方法強制 worker 退出，并從該狀態中刪除該這個worker：

export function cancelTimeout(id: string) {
 if (timeoutState[id]) {
  timeoutState[id].terminate();

  timeoutState[id] = undefined;

  return true;
 }

 return false;
}

如果你有興趣，我也實現了 setInterval，代碼在這里，但因為它對線程什么都沒做（我們重用setTimeout的代碼），所以我決定不在這里進行解釋。

我已經創建了一個短小的測試代碼，目的是檢查這種方法與原生方法的不同之處。你可以在這里找到代碼。這些是結果：

native setTimeout { ms: 7004, averageCPUCost: 0.1416 }
worker setTimeout { ms: 7046, averageCPUCost: 0.308 }

我們可以看到 setTimeout 有一點延遲 - 大約40ms - 這時 worker 被創建時的消耗。平均 CPU 成本也略高，但沒什么難以忍受的（CPU 成本是整個過程持續時間內 CPU 使用率的平均值）。

如果我們可以重用 worker，就能夠降低延遲和 CPU 使用率，這就是要實現工作池的原因。

實現工作池

如上所述，工作池是給定數量的被事先創建的 worker，他們保持空閑并監聽 message 事件。一旦 message 事件被觸發，他們就會開始工作并發回結果。

為了更好地描述我們將要做的事情，下面我們來創建一個由八個 thread worker 組成的工作池：

const pool = new WorkerPool(path.join(__dirname, './test-worker.js'), 8);

如果你熟悉限制并發操作，那么你在這里看到的邏輯幾乎相同，只是一個不同的用例。

如上面的代碼片段所示，我們把指向 worker 的路徑和要生成的 worker 數量傳給了 WorkerPool 的構造函數。

export class WorkerPool<T, N> {
 private queue: QueueItem<T, N>[] = [];
 private workersById: { [key: number]: Worker } = {};
 private activeWorkersById: { [key: number]: boolean } = {};

 public constructor(public workerPath: string, public numberOfThreads: number) {
  this.init();
 }
}

這里還有其他一些屬性，如 workersById 和 activeWorkersById，我們可以分別保存現有的 worker 和當前正在運行的 worker 的 ID。還有 queue，我們可以使用以下結構來保存對象：

type QueueCallback<N> = (err: any, result?: N) => void;

interface QueueItem<T, N> {
 callback: QueueCallback<N>;
 getData: () => T;
}

callback 只是默認的節點回調，第一個參數是錯誤，第二個參數是可能的結果。 getData 是傳遞給工作池 .run() 方法的函數（如下所述），一旦項目開始處理就會被調用。 getData 函數返回的數據將傳給工作線程。

在 .init() 方法中，我們創建了 worker 并將它們保存在以下狀態中：

private init() {
 if (this.numberOfThreads < 1) {
  return null;
 }

 for (let i = 0; i < this.numberOfThreads; i += 1) {
  const worker = new Worker(this.workerPath);

  this.workersById[i] = worker;
  this.activeWorkersById[i] = false;
 }
}

為避免無限循環，我們首先要確保線程數 > 1。然后創建有效的 worker 數，并將它們的索引保存在 workersById 狀態。我們在 activeWorkersById 狀態中保存了它們當前是否正在運行的信息，默認情況下該狀態始終為false。

現在我們必須實現前面提到的 .run() 方法來設置一個 worker 可用的任務。

public run(getData: () => T) {
 return new Promise<N>((resolve, reject) => {
  const availableWorkerId = this.getInactiveWorkerId();

  const queueItem: QueueItem<T, N> = {
   getData,
   callback: (error, result) => {
    if (error) {
     return reject(error);
    }
return resolve(result);
   },
  };

  if (availableWorkerId === -1) {
   this.queue.push(queueItem);

   return null;
  }

  this.runWorker(availableWorkerId, queueItem);
 });
}

在 promise 函數里，我們首先通過調用 .getInactiveWorkerId() 來檢查是否存在空閑的 worker 可以來處理數據：

private getInactiveWorkerId(): number {
 for (let i = 0; i < this.numberOfThreads; i += 1) {
  if (!this.activeWorkersById[i]) {
   return i;
  }
 }

 return -1;
}

接下來，我們創建一個 queueItem，在其中保存傳遞給 .run() 方法的 getData 函數以及回調。在回調中，我們要么 resolve 或者 reject promise，這取決于 worker 是否將錯誤傳遞給回調。

如果 availableWorkerId 的值是 -1，意味著當前沒有可用的 worker，我們將 queueItem 添加到 queue。如果有可用的 worker，則調用 .runWorker() 方法來執行 worker。

在 .runWorker() 方法中，我們必須把當前 worker 的 activeWorkersById 設置為使用狀態；為 message 和 error 事件設置事件監聽器（并在之后清理它們）；最后將數據發送給 worker。

private async runWorker(workerId: number, queueItem: QueueItem<T, N>) {
 const worker = this.workersById[workerId];

 this.activeWorkersById[workerId] = true;

 const messageCallback = (result: N) => {
  queueItem.callback(null, result);

  cleanUp();
 };

 const errorCallback = (error: any) => {
  queueItem.callback(error);

  cleanUp();
 };

 const cleanUp = () => {
  worker.removeAllListeners('message');
  worker.removeAllListeners('error');

  this.activeWorkersById[workerId] = false;

  if (!this.queue.length) {
   return null;
  }

  this.runWorker(workerId, this.queue.shift());
 };

 worker.once('message', messageCallback);
 worker.once('error', errorCallback);

 worker.postMessage(await queueItem.getData());
}

首先，通過使用傳遞的 workerId，我們從 workersById 中獲得 worker 引用。然后，在 activeWorkersById 中，將 [workerId] 屬性設置為true，這樣我們就能知道在 worker 在忙，不要運行其他任務。

接下來，分別創建 messageCallback 和 errorCallback 用來在消息和錯誤事件上調用，然后注冊所述函數來監聽事件并將數據發送給 worker。

在回調中，我們調用 queueItem 的回調，然后調用 cleanUp 函數。在 cleanUp 函數中，要刪除事件偵聽器，因為我們會多次重用同一個 worker。如果沒有刪除監聽器的話就會發生內存泄漏，內存會被慢慢耗盡。

在 activeWorkersById 狀態中，我們將 [workerId] 屬性設置為 false，并檢查隊列是否為空。如果不是，就從 queue 中刪除第一個項目，并用另一個 queueItem 再次調用 worker。

接著創建一個在收到 message 事件中的數據后進行一些計算的 worker：

import { isMainThread, parentPort } from 'worker_threads';

if (isMainThread) {
 throw new Error('Its not a worker');
}

const doCalcs = (data: any) => {
 const collection = [];

 for (let i = 0; i < 1000000; i += 1) {
  collection[i] = Math.round(Math.random() * 100000);
 }

 return collection.sort((a, b) => {
  if (a > b) {
   return 1;
  }

  return -1;
 });
};

parentPort.on('message', (data: any) => {
 const result = doCalcs(data);

 parentPort.postMessage(result);
});

worker 創建了一個包含 100 萬個隨機數的數組，然后對它們進行排序。只要能夠多花費一些時間才能完成，做些什么事情并不重要。

以下是工作池簡單用法的示例：

const pool = new WorkerPool<{ i: number }, number>(path.join(__dirname, './test-worker.js'), 8);

const items = [...new Array(100)].fill(null);

Promise.all(
 items.map(async (_, i) => {
  await pool.run(() => ({ i }));

  console.log('finished', i);
 }),
).then(() => {
 console.log('finished all');
});

首先創建一個由八個 worker 組成的工作池。然后創建一個包含 100 個元素的數組，對于每個元素，我們在工作池中運行一個任務。開始運行后將立即執行八個任務，其余任務被放入隊列并逐個執行。通過使用工作池，我們不必每次都創建一個 worker，從而大大提高了效率。

結論

worker_threads 提供了一種為程序添加多線程支持的簡單的方法。通過將繁重的 CPU 計算委托給其他線程，可以顯著提高服務器的吞吐量。通過官方線程支持，我們可以期待更多來自AI、機器學習和大數據等領域的開發人員和工程師使用 Node.js.

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持億速云。