該算法不復(fù)制函數(shù)、錯(cuò)誤、屬性描述符或原型鏈。還需要注意的是，以這種方式復(fù)制對(duì)象與使用 JSON 不同，因?yàn)樗梢园h(huán)引用和類型化數(shù)組，而 JSON 不能。

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

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

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

cluster 模塊可以創(chuàng)建多個(gè)節(jié)點(diǎn)實(shí)例，其中一個(gè)主進(jìn)程在它們之間對(duì)請(qǐng)求進(jìn)行路由。集群能夠有效地增加服務(wù)器的吞吐量；但是我們不能用 cluster 模塊生成一個(gè)單獨(dú)的線程。

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

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

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

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

這是一個(gè)與其父線程共享內(nèi)存的 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 });
});

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

在父線程中：

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，實(shí)際上會(huì)在兩個(gè)線程中修改它。

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

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

transferList參數(shù)

transferList 中只能包含 ArrayBuffer 和 MessagePort。一旦它們被傳送到另一個(gè)線程，就不能再次被傳送了；因?yàn)閮?nèi)存里的內(nèi)容已經(jīng)被移動(dòng)到了另一個(gè)線程。

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

創(chuàng)建通信渠道

線程之間的通信是通過 port 進(jìn)行的，port 是 MessagePort 類的實(shí)例，并啟用基于事件的通信。

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

這是一個(gè)例子：

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

parentPort.postMessage(data);

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

線程間的第二種通信方式是創(chuàng)建一個(gè) MessageChannel 并將其發(fā)送給 worker。以下代碼是如何創(chuàng)建一個(gè)新的 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]);

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

在 worker 內(nèi)部：

import { parentPort, MessagePort } from 'worker_threads';

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

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

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

使用 parentPort 不一定是錯(cuò)誤的方法，但最好用 MessageChannel 的實(shí)例創(chuàng)建一個(gè)新的 MessagePort，然后與生成的 worker 共享它。

請(qǐng)注意，在后面的例子中，為了簡(jiǎn)便起見，我用了 parentPort。

使用 worker 的兩種方式

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

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

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

這種方法被稱為工作池，因?yàn)槲覀儎?chuàng)建了一個(gè)工作池并讓它們等待，在需要時(shí)調(diào)度 message 事件來完成工作。

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

import { parentPort } from 'worker_threads';

const collection = [];

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

parentPort.postMessage(collection);

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

下面是一個(gè) worker 的例子，它可以在給定任務(wù)之前等待很長(zhǎng)一段時(shí)間：

import { parentPort } from 'worker_threads';

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

 parentPort.postMessage(result);
});

worker_threads 模塊中可用的重要屬性

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

isMainThread

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

import { isMainThread } from 'worker_threads';

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

workerData

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

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

在工作線程中：

import { workerData } from 'worker_threads';

console.log(workerData.property);

parentPort

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

threadId

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

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

實(shí)現(xiàn) setTimeout

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

import { parentPort, workerData } from 'worker_threads';

const time = Date.now();

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

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

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

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

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

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 創(chuàng)建一個(gè)唯一的標(biāo)識(shí)符，然后用先前定義的函數(shù) runWorker 來獲取 worker。我們還向 worker 傳入一個(gè)回調(diào)函數(shù)，一旦 worker 發(fā)送了數(shù)據(jù)就會(huì)被觸發(fā)。最后，把 worker 保存在狀態(tài)中并返回 id。

在回調(diào)函數(shù)中，我們必須檢查該 worker 是否仍然存在于該狀態(tài)中，因?yàn)橛锌赡軙?huì) cancelTimeout()，這將會(huì)把它刪除。如果確實(shí)存在，就把它從狀態(tài)中刪除，并調(diào)用傳給 setTimeout 函數(shù)的 callback。

cancelTimeout 函數(shù)使用 .terminate() 方法強(qiáng)制 worker 退出，并從該狀態(tài)中刪除該這個(gè)worker：

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

 timeoutState[id] = undefined;

 return true;
 }

 return false;
}

如果你有興趣，我也實(shí)現(xiàn)了 setInterval，代碼在這里，但因?yàn)樗鼘?duì)線程什么都沒做（我們重用setTimeout的代碼），所以我決定不在這里進(jìn)行解釋。

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

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

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

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

實(shí)現(xiàn)工作池

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

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

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

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

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

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，我們可以分別保存現(xiàn)有的 worker 和當(dāng)前正在運(yùn)行的 worker 的 ID。還有 queue，我們可以使用以下結(jié)構(gòu)來保存對(duì)象：

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

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

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

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

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;
 }
}

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

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

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 函數(shù)里，我們首先通過調(diào)用 .getInactiveWorkerId() 來檢查是否存在空閑的 worker 可以來處理數(shù)據(jù)：

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

 return -1;
}

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

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] 屬性設(shè)置為true，這樣我們就能知道在 worker 在忙，不要運(yùn)行其他任務(wù)。

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

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

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

接著創(chuàng)建一個(gè)在收到 message 事件中的數(shù)據(jù)后進(jìn)行一些計(jì)算的 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 創(chuàng)建了一個(gè)包含 100 萬個(gè)隨機(jī)數(shù)的數(shù)組，然后對(duì)它們進(jìn)行排序。只要能夠多花費(fèi)一些時(shí)間才能完成，做些什么事情并不重要。

以下是工作池簡(jiǎn)單用法的示例：

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');
});

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

結(jié)論

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

聲明：本網(wǎng)頁內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問題請(qǐng)及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。TEL:177 7030 7066 E-MAIL:11247931@qq.com