Node.js 10.5.0新特性-多線程

伴隨10.5.0的發佈，Node.js 新增了對多線程的實驗性支持（worker_threads模塊）。

爲何須要多線程？

Node.js因爲JS的執行在單一線程，致使CPU密集計算的任務可能會使主線程會處於繁忙的狀態，進而影響服務的性能，雖然能夠經過child_process模塊建立子進程的方式來解決，可是一方面進程之間沒法共享內存，另外一方面建立進程的開銷也不小。因此在10.5.0版本中Node.js提供了worker_threads模塊來支持多線程，一直以來被人所詬病的不擅長CPU密集計算有望成爲歷史。

如何啓用多線程？

多線程目前仍然處於實驗階段，因此啓動時須要增長--experimental-workerflag才能生效。

如何建立多線程？

worker_threads模塊中比較重要的幾個類：

MessageChannel: 用於建立異步、雙向通訊的通道實例。MessageChannel實例包含兩個屬性port1和port2，這兩個屬性都是MessagePort的實例。

MessagePort： 用於表示MessageChannel通道的終端，用於Worker之間傳輸結構化數據、內存區域和其餘的MessagePort。MessagePort繼承了EventEmitter，所以可使用postMessage和on方法實現消息的傳遞與接收。

Worker： 用於建立單獨的JS線程。

worker_threads模塊中比較重要的幾個屬性：

parentPort： 子線程中的parentPort指向能夠與主線程進行通訊的MessagePort。

子線程向父線程發送消息

parentPort.postMessage(...)
複製代碼

子線程接受來自父線程的消息

parentPort.on('message', (msg) => ...)
複製代碼

isMainThread： 用於區分當前文件是否在主線程中執行

workerData： 用於傳遞給Worker構造函數的data副本，在子線程中能夠經過workerData獲取到父進程傳入的數據。

瞭解經常使用類與屬性以後再來看一下代碼示例

const { Worker, parentPort, isMainThread } = require('worker_threads');
if (isMainThread) {
  const w = new Worker(__filename, {
    workerData: {
      name: 'Randal'
    }
  });
  w.postMessage(1e10);
  const startTime = Date.now();
  w.on('message', function(msg) {
    console.log('main thread get message: ' + msg);
    console.log('compute time ellapsed: ' + (Date.now() - startTime) / 1000);
  });
  console.log('main thread executing');
} else {
  const longComputation = (val) => {
    let sum = 0;
    for (let i = 0; i < val; i++) {
      sum += i;
    };
    return sum;
  };
  parentPort.on('message', (msg) => {
    console.log(`${workerData.name} worker get message: ` + msg);
    parentPort.postMessage(longComputation(msg));
  });
}

// 執行結果
main thread executing
Randal worker get message: 10000000000
main thread get message: 49999999990067860000
compute time ellapsed: 14.954
複製代碼

線程間如何傳輸數據？

port.postMessag(value[, transferList])
複製代碼

除了value以外，postMessage方法還支持傳入transferList參數，transferList是一個List，支持的數據類型包括ArrayBuffer和MessagePort對象，transferList中的對象在傳輸完成後，在發送對象的線程中就不能夠繼續使用了。

const { Worker, isMainThread, parentPort } = require('worker_threads');
// 主線程
if (isMainThread) {
  const sab = new ArrayBuffer(Int32Array.BYTES_PER_ELEMENT * 100);
  const ia = new Int32Array(sab);

  for (let i = 0; i < ia.length; i++) {
    ia[i] = i;
  }
  console.log("this is the main thread");
  for (let i = 0; i < 1; i++) {
    let w = new Worker(__filename);
    console.log('before transfer: ', sab);
    w.postMessage(null, [
      sab
    ]);
    setTimeout(() => {
      console.log('after transfer: ', sab);
    }, 1000);
  }
} else {
  console.log("this isn't main thread");
}
// 輸出結果
this is the main thread
before transfer:  ArrayBuffer { byteLength: 400 }
this isn't main thread after transfer: ArrayBuffer { byteLength: 0 } 複製代碼

若是ArrayBuffer是經過value傳輸的（且在transferList中不存在），則傳輸過去的是副本，以下所示：

w.postMessage(sab);

// 輸出結果
this is the main thread
before transfer:  ArrayBuffer { byteLength: 400 }
this isn't main thread after transfer: ArrayBuffer { byteLength: 400 } 複製代碼

線程間如何共享內存？

輪到SharedArrayBuffer出場了，若是postMessage中的value是SharedArrayBuffer的話，則線程之間就能夠共享內存，以下面例子所示：

const { Worker, isMainThread, parentPort } = require('worker_threads');
// 主線程
if (isMainThread) {
  const sab = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * 5);
  const ia = new Int32Array(sab);

  for (let i = 0; i < ia.length; i++) {
    ia[i] = i;
  }
  for (let i = 0; i < 2; i++) {
    let w = new Worker(__filename);
    w.postMessage(sab);
   w.on('message', () => {
    console.log(ia);
   });
  }
} else {
  parentPort.on('message', (msg) => {
    const ia = new Int32Array(msg, 0, 1);
    ia[0] = ia[0] + 1;
    parentPort.postMessage('done');
  });
}

// 輸出結果
Int32Array [ 1, 1, 2, 3, 4 ]
Int32Array [ 2, 1, 2, 3, 4 ]
複製代碼

往期回顧

Node.js線程你理解的多是錯的

Node.js child_process模塊解讀

Node.js cluster模塊解讀

參考資料

medium.com/dailyjs/thr… nodejs.org/dist/latest…