nodejs で worker_threads を使用して新しいスレッドを作成する方法

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

if (isMainThread) {
 console.log('メインスレッド内');
 新しいワーカー(__filename);
} それ以外 {
 console.log('ワーカースレッド内');
 console.log(isMainThread); // 'false' を出力します。
}

クラスMessageChannel {
  読み取り専用ポート1: MessagePort;
  読み取り専用ポート2: MessagePort;
 }

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

const { port1, port2 } = 新しい MessageChannel();
port1.on('メッセージ', (メッセージ) => console.log('受信したメッセージ', メッセージ));
port2.postMessage({ foo: 'bar' });
// 出力: `port1.on('message')` リスナーから { foo: 'bar' } を受信しました

クラスMessagePortはEventEmitterを拡張します{
  閉じる(): void;
  postMessage(値: 任意、転送リスト?: Array<ArrayBuffer | MessagePort>): void;
  ref(): void;
  参照を解除します。
  開始(): void;

  addListener(イベント: "close", リスナー: () => void): this;
  addListener(イベント: "message", リスナー: (値: any) => void): this;
  addListener(イベント: 文字列 | シンボル、リスナー: (...引数: any[]) => void): this;

  発行(イベント: "close"): ブール値;
  出力(イベント: "メッセージ", 値: 任意): boolean;
  出力(イベント: 文字列 | シンボル、...引数: any[]): boolean;

  on(イベント: "close", リスナー: () => void): this;
  on(イベント: "message", リスナー: (値: any) => void): this;
  on(イベント: 文字列 | シンボル、リスナー: (...引数: any[]) => void): this;

  一度(イベント: "close", リスナー: () => void): this;
  once(イベント: "message", リスナー: (値: any) => void): this;
  once(イベント: 文字列 | シンボル、リスナー: (...引数: any[]) => void): this;

  prependListener(イベント: "close", リスナー: () => void): this;
  prependListener(イベント: "message", リスナー: (値: any) => void): this;
  prependListener(イベント: 文字列 | シンボル、リスナー: (...引数: any[]) => void): this;

  prependOnceListener(イベント: "close", リスナー: () => void): this;
  prependOnceListener(イベント: "message", リスナー: (値: any) => void): this;
  prependOnceListener(イベント: 文字列 | シンボル、リスナー: (...引数: any[]) => void): this;

  リスナーを削除します(イベント: "close", リスナー: () => void): this;
  リスナーを削除します(イベント: "message", リスナー: (値: any) => void): this;
  リスナーを削除します(イベント: 文字列 | シンボル、リスナー: (...引数: any[]) => void): this;

  off(イベント: "close", リスナー: () => void): this;
  off(イベント: "message", リスナー: (値: any) => void): this;
  off(イベント: 文字列 | シンボル、リスナー: (...引数: any[]) => void): this;
 }

const { MessageChannel } = require('worker_threads');
const { port1, port2 } = 新しい MessageChannel();

// 印刷:
// フーバー
// 閉店しました！
port2.on('message', (message) => console.log(message));
port2.on('close', () => console.log('closed!'));

port1.postMessage('foobar');
ポート1を閉じます。

port.postMessage(値[, 転送リスト])

const { MessageChannel } = require('worker_threads');
const { port1, port2 } = 新しい MessageChannel();

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

const 円形データ = {};
円形データ.foo = 円形データ;
// 出力: { foo: [Circular] }
port2.postMessage(円形データ);

const { MessageChannel } = require('worker_threads');
const { port1, port2 } = 新しい MessageChannel();

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

const uint8Array = 新しい Uint8Array([ 1, 2, 3, 4 ]);
// uint8Array のコピーを投稿します:
port2.postMessage(uint8Array);

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

//port2.postMessage(uint8Array);

DOMException [DataCloneError]: ArrayBuffer が分離されており、複製できませんでした。

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

const pooledBuffer = 新しいArrayBuffer(8);
const typedArray1 = 新しい Uint8Array(pooledBuffer);
const typedArray2 = 新しい Float64Array(pooledBuffer);

転送不可としてマークします(pooledBuffer);

const { port1 } = 新しい MessageChannel();
port1.postMessage(typedArray1、[typedArray1.buffer]);

コンソールにログ出力します。
コンソールにログ出力します。

const { ワーカー、SHARE_ENV } = require('worker_threads');
新しいワーカー('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV })
 .on('終了', () => {
 console.log(process.env.SET_IN_WORKER); // 'foo' を出力します。
 });

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

if (isMainThread) {
 const worker = new Worker(__filename, { workerData: 'Hello, world!' });
} それ以外 {
 console.log(workerData); // 'Hello, world!' を出力します。
}

 クラスWorkerはEventEmitterを拡張します{
  読み取り専用 stdin: 書き込み可能 | null;
  読み取り専用 stdout: 読み取り可能;
  読み取り専用 stderr: 読み取り可能;
  読み取り専用スレッドID: 数値;
  読み取り専用リソース制限?: ResourceLimits;

  コンストラクター(ファイル名: 文字列 | URL、オプション: WorkerOptions);

  postMessage(値: 任意、転送リスト?: Array<ArrayBuffer | MessagePort>): void;
  ref(): void;
  参照を解除します。

  終了(): Promise<数値>;

  getHeapSnapshot(): Promise<Readable>;

  addListener(イベント: "error", リスナー: (err: Error) => void): this;
  addListener(イベント: "exit", リスナー: (exitCode: number) => void): this;
  addListener(イベント: "message", リスナー: (値: any) => void): this;
  リスナーを追加します(イベント: "online", リスナー: () => void): this;
  addListener(イベント: 文字列 | シンボル、リスナー: (...引数: any[]) => void): this;

  ... 
 }

const assert = require('assert');
定数{
 ワーカー、メッセージチャネル、メッセージポート、isMainThread、親ポート
} = 'worker_threads' が必要です。
if (isMainThread) {
 定数ワーカー = 新しいワーカー (__filename);
 定数サブチャネル = 新しいメッセージチャネル();
 ワーカー.postMessage({ hereIsYourPort: subChannel.port1 }, [subChannel.port1]);
 subChannel.port2.on('メッセージ', (値) => {
 console.log('受信:', 値);
 });
} それ以外 {
 parentPort.once('メッセージ', (値) => {
 assert(value.hereIsYourPort は MessagePort のインスタンスです);
 value.hereIsYourPort.postMessage('ワーカースレッドがこのメッセージを送信しています');
 value.hereIsYourPort.close();
 });
}

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

console.log(ポート2でメッセージを受信);
// 出力: { message: { hello: 'world' } }
console.log(ポート2でメッセージを受信);
// 出力: undefined

定数 vm = require('vm');

定数x = 1;

定数コンテキスト = { x: 2 };
vm.createContext(context); // コンテキスト分離オブジェクト。

定数コード = 'x += 40; var y = 17;';
// `x` と `y` はコンテキスト内のグローバル変数です。
// 最初は、context.x の値が 2 であるため、x の値は 2 になります。
vm.runInContext(コード、コンテキスト);

コンソールログ(コンテキストx); // 42
コンソールログ(context.y); // 17

console.log(x); // 1; yは定義されていません。

ワーカー.メッセージポートをコンテキストに移動する(ポート、コンテキスト化されたサンドボックス)

const { 親ポート } = require('worker_threads');
parentPort.on('メッセージ', (タスク) => {
 親ポートにメッセージを投稿します。
});

const { AsyncResource } = require('async_hooks');
const { EventEmitter } = require('events');
定数パス = require('path');
const { ワーカー } = require('worker_threads');

const kTaskInfo = シンボル('kTaskInfo');
const kWorkerFreedEvent = シンボル('kWorkerFreedEvent');

WorkerPoolTask​​InfoクラスはAsyncResourceを拡張します{
 コンストラクタ(コールバック) {
 スーパー('ワーカープールタスク情報');
 this.callback = コールバック;
 }

 完了(エラー、結果) {
 this.runInAsyncScope(this.callback、null、err、結果);
 this.emitDestroy(); // `TaskInfo` は 1 回だけ使用されます。
 }
}

クラス WorkerPool は EventEmitter を拡張します {
 コンストラクター(スレッド数) {
 素晴らしい（）;
 this.numThreads = numThreads;
 this.workers = [];
 this.freeWorkers = [];

 (i = 0 とします; i < numThreads; i++)
  新しいワーカーを追加します。
 }

 新しいワーカーを追加します（）{
 定数ワーカー = 新しいワーカー (path.resolve(__dirname, 'task_processor.js'));
 ワーカー.on('メッセージ', (結果) => {
  // 成功した場合: `runTask` に渡されたコールバックを呼び出します。
  // Worker に関連付けられた `TaskInfo` を削除し、空きとしてマークします
  // また。
  ワーカー[kTaskInfo].done(null、結果);
  ワーカー[kTaskInfo] = null;
  this.freeWorkers.push(ワーカー);
  this.emit(kWorkerFreedEvent);
 });
 ワーカー.on('エラー', (err) => {
  // キャッチされない例外の場合: 渡されたコールバックを呼び出します
  // エラーのある `runTask`。
  if (ワーカー[kTaskInfo])
  ワーカー[kTaskInfo].done(err, null);
  それ以外
  this.emit('エラー', err);
  // リストからワーカーを削除し、新しいワーカーを開始して置き換えます。
  // 現在のもの。
  this.workers.splice(this.workers.indexOf(ワーカー), 1);
  新しいワーカーを追加します。
 });
 this.workers.push(ワーカー);
 this.freeWorkers.push(ワーカー);
 this.emit(kWorkerFreedEvent);
 }

 runTask(タスク、コールバック) {
 if (this.freeWorkers.length === 0) {
  // 空きスレッドがないので、ワーカー スレッドが空くまで待機します。
  this.once(kWorkerFreedEvent, () => this.runTask(task, callback));
  戻る;
 }

 ワーカーを解放します。
 ワーカー[kTaskInfo] = 新しい WorkerPoolTask​​Info(コールバック);
 ワーカー.postMessage(タスク);
 }

 近い（） {
 (this.workers の const ワーカー) に対して、worker.terminate();
 }
}

モジュールをエクスポートします。

ワーカープールを require('./worker_pool.js');
os が必要です。

定数プール = 新しい WorkerPool(os.cpus().length);

終了 = 0 にします。
(i = 0; i < 10; i++ とします) {
 pool.runTask({ a: 42, b: 100 }, (err, 結果) => {
 console.log(i, err, 結果);
 (++終了 === 10)の場合
  プールを閉じます。
 });
}