nodejs中使用worker_threads來建立新的線程

時間  2021-02-17
標籤 javascript java node web 安全 異步 async 函數 oop post 欄目 JavaScript 简体版
原文   原文鏈接

[toc]javascript

nodejs中使用worker_threads來建立新的線程java

簡介

以前的文章中提到了，nodejs中有兩種線程，一種是event loop用來相應用戶的請求和處理各類callback。另外一種就是worker pool用來處理各類耗時操做。node

nodejs的官網提到了一個可以使用nodejs本地woker pool的lib叫作webworker-threads。web

惋惜的是webworker-threads的最後一次更新仍是在2年前，而在最新的nodejs 12中，根本沒法使用。安全

而webworker-threads的做者則推薦了一個新的lib叫作web-worker。異步

web-worker是構建於nodejs的worker_threads之上的，本文將會詳細講解worker_threads和web-worker的使用。async

worker_threads

worker_threads模塊的源代碼源自lib/worker_threads.js，它指的是工做線程，能夠開啓一個新的線程來並行執行javascript程序。函數

worker_threads主要用來處理CPU密集型操做,而不是IO操做，由於nodejs自己的異步IO已經很是強大了。oop

worker_threads中主要有5個屬性，3個class和3個主要的方法。接下來咱們將會一一講解。post

isMainThread

isMainThread用來判斷代碼是否在主線程中運行，咱們看一個使用的例子：

const { Worker, isMainThread } = require('worker_threads');

if (isMainThread) {
    console.log('在主線程中');
  new Worker(__filename);
} else {
  console.log('在工做線程中');
  console.log(isMainThread);  // 打印 'false'。
}

上面的例子中，咱們從worker_threads模塊中引入了Worker和isMainThread，Worker就是工做線程的主類，咱們將會在後面詳細講解，這裏咱們使用Worker建立了一個工做線程。

MessageChannel

MessageChannel表明的是一個異步雙向通訊channel。MessageChannel中沒有方法，主要經過MessageChannel來鏈接兩端的MessagePort。

class MessageChannel {
        readonly port1: MessagePort;
        readonly port2: MessagePort;
    }

當咱們使用new MessageChannel()的時候，會自動建立兩個MessagePort。

const { MessageChannel } = require('worker_threads');

const { port1, port2 } = new MessageChannel();
port1.on('message', (message) => console.log('received', message));
port2.postMessage({ foo: 'bar' });
// Prints: received { foo: 'bar' } from the `port1.on('message')` listener

經過MessageChannel，咱們能夠進行MessagePort間的通訊。

parentPort和MessagePort

parentPort是一個MessagePort類型，parentPort主要用於worker線程和主線程進行消息交互。

經過parentPort.postMessage()發送的消息在主線程中將能夠經過worker.on('message')接收。

主線程中經過worker.postMessage()發送的消息將能夠在工做線程中經過parentPort.on('message')接收。

咱們看一下MessagePort的定義：

class MessagePort extends EventEmitter {
        close(): void;
        postMessage(value: any, transferList?: Array<ArrayBuffer | MessagePort>): void;
        ref(): void;
        unref(): void;
        start(): void;

        addListener(event: "close", listener: () => void): this;
        addListener(event: "message", listener: (value: any) => void): this;
        addListener(event: string | symbol, listener: (...args: any[]) => void): this;

        emit(event: "close"): boolean;
        emit(event: "message", value: any): boolean;
        emit(event: string | symbol, ...args: any[]): boolean;

        on(event: "close", listener: () => void): this;
        on(event: "message", listener: (value: any) => void): this;
        on(event: string | symbol, listener: (...args: any[]) => void): this;

        once(event: "close", listener: () => void): this;
        once(event: "message", listener: (value: any) => void): this;
        once(event: string | symbol, listener: (...args: any[]) => void): this;

        prependListener(event: "close", listener: () => void): this;
        prependListener(event: "message", listener: (value: any) => void): this;
        prependListener(event: string | symbol, listener: (...args: any[]) => void): this;

        prependOnceListener(event: "close", listener: () => void): this;
        prependOnceListener(event: "message", listener: (value: any) => void): this;
        prependOnceListener(event: string | symbol, listener: (...args: any[]) => void): this;

        removeListener(event: "close", listener: () => void): this;
        removeListener(event: "message", listener: (value: any) => void): this;
        removeListener(event: string | symbol, listener: (...args: any[]) => void): this;

        off(event: "close", listener: () => void): this;
        off(event: "message", listener: (value: any) => void): this;
        off(event: string | symbol, listener: (...args: any[]) => void): this;
    }

MessagePort繼承自EventEmitter，它表示的是異步雙向通訊channel的一端。這個channel就叫作MessageChannel，MessagePort經過MessageChannel來進行通訊。

咱們能夠經過MessagePort來傳輸結構體數據，內存區域或者其餘的MessagePorts。

從源代碼中，咱們能夠看到MessagePort中有兩個事件，close和message。

close事件將會在channel的中任何一端斷開鏈接的時候觸發，而message事件將會在port.postMessage時候觸發，下面咱們看一個例子：

const { MessageChannel } = require('worker_threads');
const { port1, port2 } = new MessageChannel();

// Prints:
//   foobar
//   closed!
port2.on('message', (message) => console.log(message));
port2.on('close', () => console.log('closed!'));

port1.postMessage('foobar');
port1.close();

port.on('message')實際上爲message事件添加了一個listener，port還提供了addListener方法來手動添加listener。

port.on('message')會自動觸發port.start(）方法，表示啓動一個port。

當port有listener存在的時候，這表示port存在一個ref，當存在ref的時候，程序是不會結束的。咱們能夠經過調用port.unref方法來取消這個ref。

接下來咱們看一下怎麼經過port來傳輸消息：

port.postMessage(value[, transferList])

postMessage能夠接受兩個參數，第一個參數是value，這是一個JavaScript對象。第二個參數是transferList。

先看一個傳遞一個參數的狀況：

const { MessageChannel } = require('worker_threads');
const { port1, port2 } = new MessageChannel();

port1.on('message', (message) => console.log(message));

const circularData = {};
circularData.foo = circularData;
// Prints: { foo: [Circular] }
port2.postMessage(circularData);

一般來講postMessage發送的對象都是value的拷貝，可是若是你指定了transferList，那麼在transferList中的對象將會被transfer到channel的接受端，而且再也不存在於發送端，就好像把對象傳送出去同樣。

transferList是一個list，list中的對象能夠是ArrayBuffer, MessagePort 和 FileHandle。

若是value中包含SharedArrayBuffer對象，那麼該對象不能被包含在transferList中。

看一個包含兩個參數的例子：

const { MessageChannel } = require('worker_threads');
const { port1, port2 } = new MessageChannel();

port1.on('message', (message) => console.log(message));

const uint8Array = new Uint8Array([ 1, 2, 3, 4 ]);
// post uint8Array的拷貝:
port2.postMessage(uint8Array);

port2.postMessage(uint8Array, [ uint8Array.buffer ]);

//port2.postMessage(uint8Array);

上面的例子將輸出：

Uint8Array(4) [ 1, 2, 3, 4 ]
Uint8Array(4) [ 1, 2, 3, 4 ]

第一個postMessage是拷貝，第二個postMessage是transfer Uint8Array底層的buffer。

若是咱們再次調用port2.postMessage(uint8Array)，咱們會獲得下面的錯誤：

DOMException [DataCloneError]: An ArrayBuffer is detached and could not be cloned.

buffer是TypedArray的底層存儲結構，若是buffer被transfer，那麼以前的TypedArray將會變得不可用。

markAsUntransferable

要想避免這個問題，咱們能夠調用markAsUntransferable將buffer標記爲不可transferable. 咱們看一個markAsUntransferable的例子：

const { MessageChannel, markAsUntransferable } = require('worker_threads');

const pooledBuffer = new ArrayBuffer(8);
const typedArray1 = new Uint8Array(pooledBuffer);
const typedArray2 = new Float64Array(pooledBuffer);

markAsUntransferable(pooledBuffer);

const { port1 } = new MessageChannel();
port1.postMessage(typedArray1, [ typedArray1.buffer ]);

console.log(typedArray1);
console.log(typedArray2);

SHARE_ENV

SHARE_ENV是傳遞給worker構造函數的一個env變量，經過設置這個變量，咱們能夠在主線程與工做線程進行共享環境變量的讀寫。

const { Worker, SHARE_ENV } = require('worker_threads');
new Worker('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV })
  .on('exit', () => {
    console.log(process.env.SET_IN_WORKER);  // Prints 'foo'.
  });

workerData

除了postMessage()，還能夠經過在主線程中傳遞workerData給worker的構造函數，從而將主線程中的數據傳遞給worker：

const { Worker, isMainThread, workerData } = require('worker_threads');

if (isMainThread) {
  const worker = new Worker(__filename, { workerData: 'Hello, world!' });
} else {
  console.log(workerData);  // Prints 'Hello, world!'.
}

worker類

先看一下worker的定義：

class Worker extends EventEmitter {
        readonly stdin: Writable | null;
        readonly stdout: Readable;
        readonly stderr: Readable;
        readonly threadId: number;
        readonly resourceLimits?: ResourceLimits;

        constructor(filename: string | URL, options?: WorkerOptions);

        postMessage(value: any, transferList?: Array<ArrayBuffer | MessagePort>): void;
        ref(): void;
        unref(): void;

        terminate(): Promise<number>;

        getHeapSnapshot(): Promise<Readable>;

        addListener(event: "error", listener: (err: Error) => void): this;
        addListener(event: "exit", listener: (exitCode: number) => void): this;
        addListener(event: "message", listener: (value: any) => void): this;
        addListener(event: "online", listener: () => void): this;
        addListener(event: string | symbol, listener: (...args: any[]) => void): this;

       ... 
    }

worker繼承自EventEmitter，而且包含了4個重要的事件：error，exit，message和online。

worker表示的是一個獨立的 JavaScript 執行線程，咱們能夠經過傳遞filename或者URL來構造worker。

每個worker都有一對內置的MessagePort，在worker建立的時候就會相互關聯。worker使用這對內置的MessagePort來和父線程進行通訊。

經過parentPort.postMessage()發送的消息在主線程中將能夠經過worker.on('message')接收。

主線程中經過worker.postMessage()發送的消息將能夠在工做線程中經過parentPort.on('message')接收。

固然，你也能夠顯式的建立MessageChannel 對象，而後將MessagePort做爲消息傳遞給其餘線程，咱們看一個例子：

const assert = require('assert');
const {
  Worker, MessageChannel, MessagePort, isMainThread, parentPort
} = require('worker_threads');
if (isMainThread) {
  const worker = new Worker(__filename);
  const subChannel = new MessageChannel();
  worker.postMessage({ hereIsYourPort: subChannel.port1 }, [subChannel.port1]);
  subChannel.port2.on('message', (value) => {
    console.log('接收到:', value);
  });
} else {
  parentPort.once('message', (value) => {
    assert(value.hereIsYourPort instanceof MessagePort);
    value.hereIsYourPort.postMessage('工做線程正在發送此消息');
    value.hereIsYourPort.close();
  });
}

上面的例子中，咱們藉助了worker和parentPort自己的消息傳遞功能，傳遞了一個顯式的MessageChannel中的MessagePort。

而後又經過該MessagePort來進行消息的分發。

receiveMessageOnPort

除了port的on('message')方法以外，咱們還可使用receiveMessageOnPort來手動接收消息：

const { MessageChannel, receiveMessageOnPort } = require('worker_threads');
const { port1, port2 } = new MessageChannel();
port1.postMessage({ hello: 'world' });

console.log(receiveMessageOnPort(port2));
// Prints: { message: { hello: 'world' } }
console.log(receiveMessageOnPort(port2));
// Prints: undefined

moveMessagePortToContext

先了解一下nodejs中的Context的概念，咱們能夠從vm中建立context，它是一個隔離的上下文環境，從而保證不一樣運行環境的安全性，咱們看一個context的例子：

const vm = require('vm');

const x = 1;

const context = { x: 2 };
vm.createContext(context); // 上下文隔離化對象。

const code = 'x += 40; var y = 17;';
// `x` and `y` 是上下文中的全局變量。
// 最初，x 的值爲 2，由於這是 context.x 的值。
vm.runInContext(code, context);

console.log(context.x); // 42
console.log(context.y); // 17

console.log(x); // 1; y 沒有定義。

在worker中，咱們能夠將一個MessagePort move到其餘的context中。

worker.moveMessagePortToContext(port, contextifiedSandbox)

這個方法接收兩個參數，第一個參數就是要move的MessagePort,第二個參數就是vm.createContext()建立的context對象。

worker_threads的線程池

上面咱們提到了使用單個的worker thread，可是如今程序中一個線程每每是不夠的，咱們須要建立一個線程池來維護worker thread對象。

nodejs提供了AsyncResource類，來做爲對異步資源的擴展。

AsyncResource類是async_hooks模塊中的。

下面咱們看下怎麼使用AsyncResource類來建立worker的線程池。

假設咱們有一個task，使用來執行兩個數相加，腳本名字叫作task_processor.js：

const { parentPort } = require('worker_threads');
parentPort.on('message', (task) => {
  parentPort.postMessage(task.a + task.b);
});

下面是worker pool的實現：

const { AsyncResource } = require('async_hooks');
const { EventEmitter } = require('events');
const path = require('path');
const { Worker } = require('worker_threads');

const kTaskInfo = Symbol('kTaskInfo');
const kWorkerFreedEvent = Symbol('kWorkerFreedEvent');

class WorkerPoolTaskInfo extends AsyncResource {
  constructor(callback) {
    super('WorkerPoolTaskInfo');
    this.callback = callback;
  }

  done(err, result) {
    this.runInAsyncScope(this.callback, null, err, result);
    this.emitDestroy();  // `TaskInfo`s are used only once.
  }
}

class WorkerPool extends EventEmitter {
  constructor(numThreads) {
    super();
    this.numThreads = numThreads;
    this.workers = [];
    this.freeWorkers = [];

    for (let i = 0; i < numThreads; i++)
      this.addNewWorker();
  }

  addNewWorker() {
    const worker = new Worker(path.resolve(__dirname, 'task_processor.js'));
    worker.on('message', (result) => {
      // In case of success: Call the callback that was passed to `runTask`,
      // remove the `TaskInfo` associated with the Worker, and mark it as free
      // again.
      worker[kTaskInfo].done(null, result);
      worker[kTaskInfo] = null;
      this.freeWorkers.push(worker);
      this.emit(kWorkerFreedEvent);
    });
    worker.on('error', (err) => {
      // In case of an uncaught exception: Call the callback that was passed to
      // `runTask` with the error.
      if (worker[kTaskInfo])
        worker[kTaskInfo].done(err, null);
      else
        this.emit('error', err);
      // Remove the worker from the list and start a new Worker to replace the
      // current one.
      this.workers.splice(this.workers.indexOf(worker), 1);
      this.addNewWorker();
    });
    this.workers.push(worker);
    this.freeWorkers.push(worker);
    this.emit(kWorkerFreedEvent);
  }

  runTask(task, callback) {
    if (this.freeWorkers.length === 0) {
      // No free threads, wait until a worker thread becomes free.
      this.once(kWorkerFreedEvent, () => this.runTask(task, callback));
      return;
    }

    const worker = this.freeWorkers.pop();
    worker[kTaskInfo] = new WorkerPoolTaskInfo(callback);
    worker.postMessage(task);
  }

  close() {
    for (const worker of this.workers) worker.terminate();
  }
}

module.exports = WorkerPool;

咱們給worker建立了一個新的kTaskInfo屬性，而且將異步的callback封裝到WorkerPoolTaskInfo中，賦值給worker.kTaskInfo.

接下來咱們就可使用workerPool了：

const WorkerPool = require('./worker_pool.js');
const os = require('os');

const pool = new WorkerPool(os.cpus().length);

let finished = 0;
for (let i = 0; i < 10; i++) {
  pool.runTask({ a: 42, b: 100 }, (err, result) => {
    console.log(i, err, result);
    if (++finished === 10)
      pool.close();
  });
}
本文做者：flydean程序那些事

本文連接：http://www.flydean.com/nodejs-worker-thread/

本文來源：flydean的博客

歡迎關注個人公衆號:「程序那些事」最通俗的解讀，最深入的乾貨，最簡潔的教程，衆多你不知道的小技巧等你來發現！

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程