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++ 之間進行內部通信時，會導致輕微的延遲，但這幾乎不可察覺。

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

1fs.readFile(path.join(__dirname, './package.json'), (err, content) => {
2 if (err) {
3 return null;
4 }
5
6 console.log(content.toString());
7});

前面提到的 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，然後再討論它們的屬性。

 1type WorkerCallback = (err: any, result?: any) => any;
 2
 3export function runWorker(path: string, cb: WorkerCallback, workerData: object | null = null) {
 4 const worker = new Worker(path, { workerData });
 5
 6 worker.on('message', cb.bind(null, null));
 7 worker.on('error', cb);
 8
 9 worker.on('exit', (exitCode) => {
10 if (exitCode === 0) {
11 return null;
12 }
13
14 return cb(new Error(`Worker has stopped with code ${exitCode}`));
15 });
16
17 return worker;
18}

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

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

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

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

以下是最常見的事件：

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

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

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

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

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

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

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

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

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

在線程之間交換數據

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

1port.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：

 1import { parentPort } from 'worker_threads';
 2
 3parentPort.on('message', () => {
 4 const numberOfElements = 100;
 5 const sharedBuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * numberOfElements);
 6 const arr = new Int32Array(sharedBuffer);
 7
 8 for (let i = 0; i < numberOfElements; i += 1) {
 9 arr[i] = Math.round(Math.random() * 30);
10 }
11
12 parentPort.postMessage({ arr });
13});

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

在父線程中：

 1import path from 'path';
 2
 3import { runWorker } from '../run-worker';
 4
 5const worker = runWorker(path.join(__dirname, 'worker.js'), (err, { arr }) => {
 6 if (err) {
 7 return null;
 8 }
 9
10 arr[0] = 5;
11});
12
13worker.postMessage({});

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

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

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

transferList參數

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

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

創建通信渠道

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

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

這是一個例子：

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

4};
5
6parentPort.postMessage(data);

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

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

 1import path from 'path';
 2import { Worker, MessageChannel } from 'worker_threads';
 3
 4const worker = new Worker(path.join(__dirname, 'worker.js'));
 5
 6const { port1, port2 } = new MessageChannel();
 7
 8port1.on('message', (message) => {
 9 console.log('message from worker:', message);
10});
11
12worker.postMessage({ port: port2 }, [port2]);

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

在 worker 內部：

1import { parentPort, MessagePort } from 'worker_threads';
2
3parentPort.on('message', (data) => {
4 const { port }: { port: MessagePort } = data;
5
6 port.postMessage('heres your message!');
7});

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

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

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

使用 worker 的兩種方式

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

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

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

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

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

1import { parentPort } from 'worker_threads';
2
3const collection = [];
4
5for (let i = 0; i < 10; i += 1) {
6 collection[i] = i;
7}
8
9parentPort.postMessage(collection);

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

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

1import { parentPort } from 'worker_threads';
2
3parentPort.on('message', (data: any) => {
4 const result = doSomething(data);
5
6 parentPort.postMessage(result);
7});

worker_threads 模塊中可用的重要屬性

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

isMainThread

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

1import { isMainThread } from 'worker_threads';
2
3if (isMainThread) { 

4 throw new Error('Its not a worker');
5}

workerData

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

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

在工作線程中：

1import { workerData } from 'worker_threads';
2
3console.log(workerData.property);

parentPort

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

threadId

分配給 worker 的唯一標識符。

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

實現 setTimeout

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

 1import { parentPort, workerData } from 'worker_threads';
 2
 3const time = Date.now();
 4
 5while (true) {
 6 if (time + workerData.time <= Date.now()) {
 7 parentPort.postMessage({});
 8 break;
 9 }
10}

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

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

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

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

 1export function setTimeout(callback: (err: any) => any, time: number) {
 2 const id = uuidv4();
 3
 4 const worker = runWorker(
 5 path.join(__dirname, './timeout-worker.js'),
 6 (err) => {
 7 if (!timeoutState[id]) {
 8 return null;
 9 }
10
11 timeoutState[id] = null; 

12
13 if (err) {
14 return callback(err);
15 }
16
17 callback(null);
18 },
19 {
20 time,
21 },
22 );
23
24 timeoutState[id] = worker;
25
26 return id;
27}

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

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

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

 1export function cancelTimeout(id: string) {
 2 if (timeoutState[id]) {
 3 timeoutState[id].terminate();
 4
 5 timeoutState[id] = undefined;
 6
 7 return true;
 8 }
 9
10 return false; 

11}

如果你有興趣，我也實現了 setInterval，代碼在這裡【https://github.com/maciejcieslar/threads-nodejs/blob/master/src/timeout/timeout.ts#L64】，但因為它對線程什麼都沒做（我們重用setTimeout的代碼），所以我決定不在這裡進行解釋。

我已經創建了一個短小的測試代碼，目的是檢查這種方法與原生方法的不同之處。你可以在這裡找到代碼【https://github.com/maciejcieslar/threads-nodejs/blob/master/src/timeout/index.ts#L13】。這些是結果：

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

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

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

實現工作池

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

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

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

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

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

1export class WorkerPool {
2 private queue: QueueItem[] = [];
3 private workersById: { [key: number]: Worker } = {};
4 private activeWorkersById: { [key: number]: boolean } = {};
5
6 public constructor(public workerPath: string, public numberOfThreads: number) {
7 this.init();
8 }
9}

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

1type QueueCallback = (err: any, result?: N) => void;
2
3interface QueueItem {
4 callback: QueueCallback;
5 getData: () => T;
6}
 

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

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

 1private init() {
 2 if (this.numberOfThreads < 1) {
 3 return null;
 4 }
 5
 6 for (let i = 0; i < this.numberOfThreads; i += 1) {
 7 const worker = new Worker(this.workerPath);
 8
 9 this.workersById[i] = worker;
10 this.activeWorkersById[i] = false;
11 }
12}

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

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

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

 4
 5 const queueItem: QueueItem = {
 6 getData,
 7 callback: (error, result) => {
 8 if (error) {
 9 return reject(error);
10 }
11return resolve(result);
12 },
13 };
14
15 if (availableWorkerId === -1) {
16 this.queue.push(queueItem);
17
18 return null;
19 }
20
21 this.runWorker(availableWorkerId, queueItem);
22 });
23}

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

1private getInactiveWorkerId(): number {
2 for (let i = 0; i < this.numberOfThreads; i += 1) {
3 if (!this.activeWorkersById[i]) {
4 return i;
5 }
6 }
7
8 return -1;
9}

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

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

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

 1private async runWorker(workerId: number, queueItem: QueueItem) {
 2 const worker = this.workersById[workerId];
 3
 4 this.activeWorkersById[workerId] = true;
 5
 6 const messageCallback = (result: N) => {
 7 queueItem.callback(null, result);
 8
 9 cleanUp();
10 };
11
12 const errorCallback = (error: any) => {
13 queueItem.callback(error);
14
15 cleanUp();
16 };
17
18 const cleanUp = () => {
19 worker.removeAllListeners('message');
20 worker.removeAllListeners('error');
21
22 this.activeWorkersById[workerId] = false;
23
24 if (!this.queue.length) {
25 return null;
26 }
27
28 this.runWorker(workerId, this.queue.shift());
29 };
30
31 worker.once('message', messageCallback);
32 worker.once('error', errorCallback);
33
34 worker.postMessage(await queueItem.getData());
35}

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

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

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

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

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

 1import { isMainThread, parentPort } from 'worker_threads';
 2
 3if (isMainThread) {
 4 throw new Error('Its not a worker');
 5}
 6
 7const doCalcs = (data: any) => {
 8 const collection = [];
 9
10 for (let i = 0; i < 1000000; i += 1) {
11 collection[i] = Math.round(Math.random() * 100000);
12 }
13
14 return collection.sort((a, b) => {
15 if (a > b) {
16 return 1;
17 }
18
19 return -1;
20 });
21};
22 

23parentPort.on('message', (data: any) => {
24 const result = doCalcs(data);
25
26 parentPort.postMessage(result);
27});

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

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

 1const pool = new WorkerPool(path.join(__dirname, './test-worker.js'), 8);
 2
 3const items = [...new Array(100)].fill(null);
 4
 5Promise.all(
 6 items.map(async (_, i) => {
 7 await pool.run(() => ({ i }));
 8
 9 console.log('finished', i);
10 }),
11).then(() => {
12 console.log('finished all');
13});

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

結論

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

十五年編程經驗，今年1月整理了一批2019年最新WEB前端教學視頻，不論是零基礎想要學習前端還是學完在工作想要提升自己，這些資料都會給你帶來幫助，從HTML到各種框架，幫助所有想要學好前端的同學，學習規劃、學習路線、學習資料、問題解答。只要關注我的頭條號，後臺私信我【前端】兩個字，即可免費獲取。

閱讀更多 前端技術教學分享 的文章

關鍵字: 中央處理器 單線程 設計