自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

Node.js多線程完全指南

開發(fā) 前端
很多人都想知道單線程的 Node.js 怎么能與多線程后端競爭??紤]到其所謂的單線程特性,許多大公司選擇 Node 作為其后端似乎違反直覺。要想知道原因,必須理解其單線程的真正含義。

[[260976]]

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

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

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

Node.js 是如何工作的

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

事件循環(huán)是一種機制,它采用回調(diào)(函數(shù))并注冊它們,準(zhǔn)備在將來的某個時刻執(zhí)行。它與相關(guān)的 JavaScript 代碼在同一個線程中運行。當(dāng) JavaScript 操作阻塞線程時,事件循環(huán)也會被阻止。

工作池是一種執(zhí)行模型,它產(chǎn)生并處理單獨的線程,然后同步執(zhí)行任務(wù),并將結(jié)果返回到事件循環(huán)。事件循環(huán)使用返回的結(jié)果執(zhí)行提供的回調(diào)。

簡而言之,它負責(zé)異步 I/O操作 —— 主要是與系統(tǒng)磁盤和網(wǎng)絡(luò)的交互。它主要由諸如 fs(I/O 密集)或 crypto(CPU 密集)等模塊使用。工作池用 libuv 實現(xiàn),當(dāng) Node 需要在 JavaScript 和 C++ 之間進行內(nèi)部通信時,會導(dǎo)致輕微的延遲,但這幾乎不可察覺。

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

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

前面提到的 fs 模塊告訴工作池使用其中一個線程來讀取文件的內(nèi)容,并在完成后通知事件循環(huán)。然后事件循環(huán)獲取提供的回調(diào)函數(shù),并用文件的內(nèi)容執(zhí)行它。

以上是非阻塞代碼的示例,我們不必同步等待某事的發(fā)生。只需告訴工作池去讀取文件,并用結(jié)果去調(diào)用提供的函數(shù)即可。由于工作池有自己的線程,因此事件循環(huán)可以在讀取文件時繼續(xù)正常執(zhí)行。

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

在需要對數(shù)據(jù)進行復(fù)雜的計算時(如AI、機器學(xué)習(xí)或大數(shù)據(jù))無法真正有效地使用 Node.js,因為操作阻塞了主(且唯一)線程,使服務(wù)器無響應(yīng)。在 Node.js v10.5.0 發(fā)布之前就是這種情況,在這一版本增加了對多線程的支持。

簡介:worker_threads

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

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

注意,術(shù)語 thread worker,worker 和 thread 經(jīng)常互換使用,他們都指的是同一件事。

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

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

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

請注意:不管你是用的是 JavaScript, 還是最終要轉(zhuǎn)換為 JavaScript 的語言(例如,TypeScript),路徑應(yīng)該始終引用帶有 .js 或 .mjs 擴展名的文件。

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

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

以下是最常見的事件: 

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

只要 worker 中有未捕獲的異常,就會發(fā)出 error 事件。然后終止 worker,錯誤可以作為提供的回調(diào)中的***個參數(shù)。 

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

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

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

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

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

只要 worker 將數(shù)據(jù)發(fā)送到父線程,就會發(fā)出 message 事件。

現(xiàn)在讓我們來看看如何在線程之間共享數(shù)據(jù)。

在線程之間交換數(shù)據(jù)

要將數(shù)據(jù)發(fā)送到另一個線程,可以用 port.postMessage() 方法。它的原型如下: 

  1. port.postMessage(data[, transferList]) 

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

數(shù)據(jù)參數(shù)

***個參數(shù) —— 這里被稱為 data —— 是一個被復(fù)制到另一個線程的對象。它可以是復(fù)制算法所支持的任何內(nèi)容。

數(shù)據(jù)由結(jié)構(gòu)化克隆算法進行復(fù)制。引用自 Mozilla:

它通過遞歸輸入對象來進行克隆,同時保持之前訪問過的引用的映射,以避免***遍歷循環(huán)。

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

由于能夠復(fù)制類型化數(shù)組,該算法可以在線程之間共享內(nèi)存。

在線程之間共享內(nèi)存

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

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

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

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

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

現(xiàn)在讓我們看一下如何在線程之間共享內(nèi)存。為了共享內(nèi)存,必須將 ArrayBuffer 或 SharedArrayBuffer 的實例作為數(shù)據(jù)參數(shù)發(fā)送到另一個線程。

這是一個與其父線程共享內(nèi)存的 worker: 

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

首先,我們創(chuàng)建一個 SharedArrayBuffer,其內(nèi)存需要包含100個32位整數(shù)。接下來創(chuàng)建一個 Int32Array 實例,它將用緩沖區(qū)來保存其結(jié)構(gòu),然后用一些隨機數(shù)填充數(shù)組并將其發(fā)送到父線程。

在父線程中: 

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

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

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

共享一個整數(shù)數(shù)組固然很好,但我們真正感興趣的是共享對象 —— 這是存儲信息的默認方式。不幸的是,沒有 SharedObjectBuffer 或類似的東西,但我們可以自己創(chuàng)建一個類似的結(jié)構(gòu)。

transferList參數(shù)

transferList 中只能包含 ArrayBuffer 和 MessagePort。一旦它們被傳送到另一個線程,就不能再次被傳送了;因為內(nèi)存里的內(nèi)容已經(jīng)被移動到了另一個線程。

目前,還不能通過 transferList(可以使用 child_process 模塊)來傳輸網(wǎng)絡(luò)套接字。

創(chuàng)建通信渠道

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

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

這是一個例子: 

  1. import { parentPort } from 'worker_threads';  
  2. const data = {  
  3.  // ...  
  4. };  
  5. parentPort.postMessage(data); 

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

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

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

在創(chuàng)建 port1 和 port2 之后,我們在 port1 上設(shè)置事件監(jiān)聽器并將 port2 發(fā)送給 worker。我們必須將它包含在 transferList 中,以便將其傳輸給 worker 。

在 worker 內(nèi)部: 

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

這樣,我們就能使用父線程發(fā)送的 port 了。

使用 parentPort 不一定是錯誤的方法,但***用 MessageChannel 的實例創(chuàng)建一個新的 MessagePort,然后與生成的 worker 共享它。

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

使用 worker 的兩種方式

可以通過兩種方式使用 worker。***種是生成一個 worker,然后執(zhí)行它的代碼,并將結(jié)果發(fā)送到父線程。通過這種方法,每當(dāng)出現(xiàn)新任務(wù)時,都必須重新創(chuàng)建一個工作者。

第二種方法是生成一個 worker 并為 message 事件設(shè)置監(jiān)聽器。每次觸發(fā) message 時,它都會完成工作并將結(jié)果發(fā)送回父線程,這會使 worker 保持活動狀態(tài)以供以后使用。

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

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

以下是一個產(chǎn)生、執(zhí)行然后關(guān)閉 worker 例子: 

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

將 collection 發(fā)送到父線程后,它就會退出。

下面是一個 worker 的例子,它可以在給定任務(wù)之前等待很長一段時間: 

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

worker_threads 模塊中可用的重要屬性

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

isMainThread

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

  1. import { isMainThread } from 'worker_threads';  
  2. if (isMainThread) {  
  3.  throw new Error('Its not a worker');  

workerData

產(chǎn)生線程時包含在 worker 的構(gòu)造函數(shù)中的數(shù)據(jù)。 

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

在工作線程中: 

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

parentPort

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

threadId

分配給 worker 的唯一標(biāo)識符。

現(xiàn)在我們知道了技術(shù)細節(jié),接下來實現(xiàn)一些東西并在實踐中檢驗學(xué)到的知識。

實現(xiàn) setTimeout

setTimeout 是一個***循環(huán),顧名思義,用來檢測程序運行時間是否超時。它在循環(huán)中檢查起始時間與給定毫秒數(shù)之和是否小于實際日期。 

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

這個特定的實現(xiàn)產(chǎn)生一個線程,然后執(zhí)行它的代碼,***在完成后退出。

接下來實現(xiàn)使用這個 worker 的代碼。首先創(chuàng)建一個狀態(tài),用它來跟蹤生成的 worker: 

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

然后時負責(zé)創(chuàng)建 worker 并將其保存到狀態(tài)的函數(shù): 

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

首先,我們使用 UUID 包為 worker 創(chuàng)建一個唯一的標(biāo)識符,然后用先前定義的函數(shù) runWorker 來獲取 worker。我們還向 worker 傳入一個回調(diào)函數(shù),一旦 worker 發(fā)送了數(shù)據(jù)就會被觸發(fā)。***,把 worker 保存在狀態(tài)中并返回 id。

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

cancelTimeout 函數(shù)使用 .terminate() 方法強制 worker 退出,并從該狀態(tài)中刪除該這個worker: 

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

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

我已經(jīng)創(chuàng)建了一個短小的測試代碼,目的是檢查這種方法與原生方法的不同之處。你可以在這里找到代碼。這些是結(jié)果: 

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

我們可以看到 setTimeout 有一點延遲 - 大約40ms - 這時 worker 被創(chuàng)建時的消耗。平均 CPU 成本也略高,但沒什么難以忍受的(CPU 成本是整個過程持續(xù)時間內(nèi) CPU 使用率的平均值)。

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

實現(xiàn)工作池

如上所述,工作池是給定數(shù)量的被事先創(chuàng)建的 worker,他們保持空閑并監(jiān)聽 message 事件。一旦 message 事件被觸發(fā),他們就會開始工作并發(fā)回結(jié)果。

為了更好地描述我們將要做的事情,下面我們來創(chuàng)建一個由八個 thread worker 組成的工作池: 

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

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

如上面的代碼片段所示,我們把指向 worker 的路徑和要生成的 worker 數(shù)量傳給了 WorkerPool 的構(gòu)造函數(shù)。 

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

這里還有其他一些屬性,如 workersById 和 activeWorkersById,我們可以分別保存現(xiàn)有的 worker 和當(dāng)前正在運行的 worker 的 ID。還有 queue,我們可以使用以下結(jié)構(gòu)來保存對象: 

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

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

在 .init() 方法中,我們創(chuàng)建了 worker 并將它們保存在以下狀態(tài)中: 

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

為避免***循環(huán),我們首先要確保線程數(shù) > 1。然后創(chuàng)建有效的 worker 數(shù),并將它們的索引保存在 workersById 狀態(tài)。我們在 activeWorkersById 狀態(tài)中保存了它們當(dāng)前是否正在運行的信息,默認情況下該狀態(tài)始終為false。

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

  1. public run(getData: () => T) {  
  2.   return new Promise<N>((resolve, reject) => {  
  3.     const availableWorkerId = this.getInactiveWorkerId();  
  4.     const queueItem: QueueItem<T, N> = {  
  5.       getData,  
  6.       callback: (error, result) => {  
  7.         if (error) {  
  8.           return reject(error);  
  9.         }  
  10. return resolve(result);  
  11.       },  
  12.     };  
  13.     if (availableWorkerId === -1) {  
  14.       this.queue.push(queueItem);  
  15.       return null;  
  16.     }  
  17.     this.runWorker(availableWorkerId, queueItem);  
  18.   });  

在 promise 函數(shù)里,我們首先通過調(diào)用 .getInactiveWorkerId() 來檢查是否存在空閑的 worker 可以來處理數(shù)據(jù): 

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

接下來,我們創(chuàng)建一個 queueItem,在其中保存?zhèn)鬟f給 .run() 方法的 getData 函數(shù)以及回調(diào)。在回調(diào)中,我們要么 resolve 或者 reject promise,這取決于 worker 是否將錯誤傳遞給回調(diào)。

如果 availableWorkerId 的值是 -1,意味著當(dāng)前沒有可用的 worker,我們將 queueItem 添加到 queue。如果有可用的 worker,則調(diào)用 .runWorker() 方法來執(zhí)行 worker。

在 .runWorker() 方法中,我們必須把當(dāng)前 worker 的 activeWorkersById 設(shè)置為使用狀態(tài);為 message 和 error 事件設(shè)置事件監(jiān)聽器(并在之后清理它們);***將數(shù)據(jù)發(fā)送給 worker。 

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

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

接下來,分別創(chuàng)建 messageCallback 和 errorCallback 用來在消息和錯誤事件上調(diào)用,然后注冊所述函數(shù)來監(jiān)聽事件并將數(shù)據(jù)發(fā)送給 worker。

在回調(diào)中,我們調(diào)用 queueItem 的回調(diào),然后調(diào)用 cleanUp 函數(shù)。在 cleanUp 函數(shù)中,要刪除事件偵聽器,因為我們會多次重用同一個 worker。如果沒有刪除監(jiān)聽器的話就會發(fā)生內(nèi)存泄漏,內(nèi)存會被慢慢耗盡。

在 activeWorkersById 狀態(tài)中,我們將 [workerId] 屬性設(shè)置為 false,并檢查隊列是否為空。如果不是,就從 queue 中刪除***個項目,并用另一個 queueItem 再次調(diào)用 worker。

接著創(chuàng)建一個在收到 message 事件中的數(shù)據(jù)后進行一些計算的 worker: 

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

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

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

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

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

結(jié)論

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

 

責(zé)任編輯:龐桂玉 來源: segmyntfault
相關(guān)推薦

2011-11-10 08:55:00

Node.js

2015-03-10 10:59:18

Node.js開發(fā)指南基礎(chǔ)介紹

2021-04-20 12:39:52

Node.js多線程多進程

2021-08-04 23:30:28

Node.js開發(fā)線程

2021-02-01 15:42:45

Node.jsSQL應(yīng)用程序

2021-08-24 05:00:21

Nodejs線程

2015-06-30 08:41:55

Node.js指南

2021-05-21 09:36:42

開發(fā)技能代碼

2021-03-09 08:03:21

Node.js 線程JavaScript

2014-08-01 09:57:52

Node.jsNode.js插件

2013-11-01 09:34:56

Node.js技術(shù)

2022-06-23 06:34:56

Node.js子線程

2020-09-28 06:57:39

Node.jsGraphQLAPI

2011-09-09 14:23:13

Node.js

2011-11-01 10:30:36

Node.js

2011-09-08 13:46:14

node.js

2011-09-02 14:47:48

Node

2012-10-24 14:56:30

IBMdw

2022-01-29 22:27:31

內(nèi)核子線程應(yīng)用

2021-08-25 06:33:52

Node.jsVscode調(diào)試工具
點贊
收藏

51CTO技術(shù)棧公眾號