Node.js中怎么實現多線程

Node.js中怎么實現多線程，針對這個問題，這篇文章詳細介紹了相對應的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

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 | nullnull = 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，從而大大提高了效率。

關于Node.js中怎么實現多線程問題的解答就分享到這里了，希望以上內容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關注億速云行業資訊頻道了解更多相關知識。