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

實現(xiàn) setTimeout

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

import { parentPort, workerData } from 'worker_threads';

const time = Date.now();

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

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

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

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

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

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

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

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

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

   timeoutState[id] = undefined;

   return true;
 }

 return false;
}

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

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

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

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

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

實現(xiàn)工作池

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

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

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

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

如上面的代碼片段所示，我們把指向 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，我們可以分別保存現(xiàn)有的 worker 和當前正在運行的 worker 的 ID。還有 queue，我們可以使用以下結構來保存對象：

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

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

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

在 .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)，我們首先要確保線程數 > 1。然后創(chuàng)建有效的 worker 數，并將它們的索引保存在 workersById 狀態(tài)。我們在 activeWorkersById 狀態(tài)中保存了它們當前是否正在運行的信息，默認情況下該狀態(tài)始終為false。

現(xiàn)在我們必須實現(xiàn)前面提到的 .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;
}

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

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

在 .runWorker() 方法中，我們必須把當前 worker 的 activeWorkersById 設置為使用狀態(tài)；為 message 和 error 事件設置事件監(jiān)聽器（并在之后清理它們）；最后將數據發(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] 屬性設置為true，這樣我們就能知道在 worker 在忙，不要運行其他任務。

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

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

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

接著創(chuàng)建一個在收到 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 創(chuàng)建了一個包含 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');
});

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

結論

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

英文原文地址：https://blog.logrocket.com/a-complete-guide-to-threads-in-node-js-4fa3898fe74f

相關推薦：nodejs 教程

更多關于云服務器，域名注冊，虛擬主機的問題，請訪問西部數碼官網：m.ps-sw.cn





猜你還會喜歡下面的內容

• 做一個獨立站需要多少錢
• 西部數碼怎樣搶注域名
• B2C 電子商務網站如何做推廣和優(yōu)化
• 網站推廣增加外鏈最好的方法
• 新手站長淺談：如何在線提升產品轉化率
• 網站策劃：九大思維理念助你策劃能力提升
• 如何高效輸出移動app產品原型？
• 為什么說產品思考是進行用戶體驗設計時所不可或缺的？
• 六套 App：構建我的產品設計工作流
• 如果Vine模式重來，大概就是Biu這個樣子
• 新手入門：如何做需求分析
• 產品狗的日常：如何更好的去分解功能點
• 簡單明了丨如何寫出好的PRD
• 產品必修課——輕松搞定UML
• 如果你是個苦逼的App測試人員，這款產品或許能拯救你
• 手把手教你制作實用美觀的交互設計文檔

熱門標簽

其他問題云服務器問題域名及賬戶問題企業(yè)郵局市場咨詢云服務器云建站/云站群/小程序虛擬主機網絡知識企業(yè)郵箱域名注冊域名備案域名商標注冊云主機更多標簽...

大家感興趣的內容