ワーカースレッド

1. 学習内容

Worker クラスを使用して、ワーカースレッドを作成し、そのライフサイクルを管理する
parentPort を使用して、メインスレッドとワーカー間の双方向メッセージ通信を実装する
workerData を使用して、初期化データをワーカーに渡す
MessageChannel / MessagePort を使用して、スレッド間の直接通信を確立する
SharedArrayBuffer および Atomics を使用して、共有メモリ操作を実行する
ワーカーを再利用し、スループットを向上させるためのスレッドプールパターンを構築する
ワーカースレッド、child_process、およびclusterの適切な使用例を比較する

2. ストーリー：アリスの画像処理の高速化

アリスは、あるSaaS企業で画像処理サービスの責任者を務めています。ユーザーが画像をアップロードすると、システムは各画像について3種類のサイズのサムネイルを生成する必要があります。当初のシングルスレッド方式のソリューションでは、100枚の画像を処理するのに約30秒かかり、ピーク時にはリクエストが大幅に滞る事態が発生していました。

彼女は、ワーカースレッドを使用してタスクを4つのスレッドに分散させ、並列処理を行うことにした。メインスレッドがファイル一覧を読み込み、workerDataを介してタスクのパラメータを渡すと、ワーカースレッドはサムネイルを生成した後、parentPortを介して結果を返す。その結果、100枚の画像の処理時間は約8秒に短縮され、スループットは4倍近く向上した。

3. ワーカークラスの基礎

worker_threads モジュールは、Node.js で真のマルチスレッド機能を提供します。各 Worker は、別々のスレッドでスクリプトを実行し、独自の V8 エンジンインスタンスとイベントループを持っています。

JAVASCRIPT

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

const worker = new Worker('./heavy-task.js', {
  workerData: { taskId: 42, input: 'hello' }
});

worker.on('message', (result) => {
  console.log('Result from worker:', result);
});

worker.on('error', (err) => {
  console.error('Worker error:', err);
});

worker.on('exit', (code) => {
  if (code !== 0) {
    console.error(`Worker stopped with exit code ${code}`);
  }
});

「Worker」スクリプト内部：

JAVASCRIPT

// heavy-task.js
const { parentPort, workerData } = require('worker_threads');

const { taskId, input } = workerData;

const result = performHeavyComputation(input);

parentPort.postMessage({ taskId, result });

function performHeavyComputation(data) {
  let hash = 0;
  for (let i = 0; i < 1e8; i++) {
    hash = (hash + data.charCodeAt(i % data.length)) % 65536;
  }
  return hash;
}

▶ 例：基本的なワーカーの作成と通信

JAVASCRIPT

// main.js
const { Worker } = require('worker_threads');

const worker = new Worker(`
  const { parentPort } = require('worker_threads');
  parentPort.postMessage('Hello from worker!');
`, { eval: true });

worker.on('message', (msg) => {
  console.log(msg); // Hello from worker!
});

4. parentPort 双方向通信

parentPort は、ワーカーとメインスレッド間のメッセージチャネルです。メインスレッドは worker.postMessage() 経由でメッセージを送信し、ワーカーは parentPort.postMessage() 経由で応答します。双方は、データを受信するために message イベントをリッスンしています。

▶ 例：双方向のメッセージ交換

JAVASCRIPT

// main.js
const { Worker } = require('worker_threads');

const worker = new Worker('./echo-worker.js');

worker.postMessage({ action: 'greet', name: 'Alice' });

worker.on('message', (msg) => {
  console.log('Main received:', msg);
  if (msg.action === 'greet_reply') {
    worker.postMessage({ action: 'task', payload: 100 });
  }
});

JAVASCRIPT

// echo-worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', (msg) => {
  if (msg.action === 'greet') {
    parentPort.postMessage({
      action: 'greet_reply',
      text: `Hello, ${msg.name}!`
    });
  } else if (msg.action === 'task') {
    const result = msg.payload * 2;
    parentPort.postMessage({ action: 'result', value: result });
  }
});

▶ 例：転送可能なオブジェクトのゼロコピー転送

JAVASCRIPT

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

const buffer = new ArrayBuffer(1024 * 1024); // 1 MB

const worker = new Worker('./process-buffer.js');

worker.postMessage({ buffer }, [buffer]);

console.log('Buffer transferred, main thread no longer owns it');

JAVASCRIPT

// process-buffer.js
const { parentPort, workerData } = require('worker_threads');

parentPort.on('message', ({ buffer }) => {
  const view = new Uint8Array(buffer);
  view[0] = 42;
  parentPort.postMessage({ done: true, firstByte: view[0] }, [buffer]);
});

5. workerData の初期化データ

workerData は、ワーカーの作成時にオプションを介して渡される読み取り専用の初期データです。ワーカーは、メッセージのやり取りを必要とせずに、内部でこのデータを直接読み取ります。設定、ファイルパス、タスクのパラメータなどを渡すのに適しています。

▶ 例：画像バッチ処理ワーカー

JAVASCRIPT

// main.js
const { Worker } = require('worker_threads');
const path = require('path');

const imageFiles = [
  'photo-001.jpg', 'photo-002.jpg', 'photo-003.jpg',
  'photo-004.jpg', 'photo-005.jpg', 'photo-006.jpg'
];

const WORKER_COUNT = 4;
const chunkSize = Math.ceil(imageFiles.length / WORKER_COUNT);

for (let i = 0; i < WORKER_COUNT; i++) {
  const chunk = imageFiles.slice(i * chunkSize, (i + 1) * chunkSize);
  const worker = new Worker('./image-worker.js', {
    workerData: {
      workerId: i,
      files: chunk,
      outputDir: './thumbnails'
    }
  });

  worker.on('message', (result) => {
    console.log(`Worker ${i} done:`, result.processed);
  });
}

JAVASCRIPT

// image-worker.js
const { parentPort, workerData } = require('worker_threads');
const path = require('path');

const { workerId, files, outputDir } = workerData;

async function generateThumbnail(file) {
  // Simulate image processing
  return new Promise((resolve) => {
    setTimeout(() => resolve(`${file} -> thumb`), 100);
  });
}

(async () => {
  const processed = [];
  for (const file of files) {
    const result = await generateThumbnail(file);
    processed.push(result);
  }
  parentPort.postMessage({ workerId, processed });
})();

6. MessageChannel と MessagePort

MessageChannel 異なるワーカーに割り当て可能な、相互に接続された MessagePort インスタンスのペアを作成し、メインスレッドを経由せずにスレッド間で直接通信できるようにします。

graph LR
    Main["Main Thread"] -->|"worker.postMessage()"| W1["Worker 1"]
    W1 -->|"parentPort.postMessage()"| Main
    Main -->|"worker.postMessage()"| W2["Worker 2"]
    W2 -->|"parentPort.postMessage()"| Main
    W1 <-->|"MessagePort"| W2

▶ 例：2人の作業員が直接やり取りする場合

JAVASCRIPT

// main.js
const { Worker, MessageChannel } = require('worker_threads');

const worker1 = new Worker('./chat-worker.js', {
  workerData: { id: 1 }
});
const worker2 = new Worker('./chat-worker.js', {
  workerData: { id: 2 }
});

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

worker1.postMessage({ port: port1 }, [port1]);
worker2.postMessage({ port: port2 }, [port2]);

JAVASCRIPT

// chat-worker.js
const { parentPort, workerData } = require('worker_threads');

parentPort.once('message', ({ port }) => {
  port.on('message', (msg) => {
    console.log(`Worker ${workerData.id} received:`, msg);
  });

  setInterval(() => {
    port.postMessage(`Hello from Worker ${workerData.id}`);
  }, 1000);
});

7. SharedArrayBuffer とアトミック操作

SharedArrayBuffer 複数のスレッドが同じメモリブロックを共有できるようにします。Atomics が提供するアトミック操作と組み合わせて使用することで、共有データの安全な読み書きが可能となり、レースコンディションを回避できます。

▶ 例：共有カウンター

JAVASCRIPT

// main.js
const { Worker } = require('worker_threads');

const sharedBuffer = new SharedArrayBuffer(4);
const sharedArray = new Int32Array(sharedBuffer);

const WORKER_COUNT = 4;
const INCREMENTS_PER_WORKER = 100000;

for (let i = 0; i < WORKER_COUNT; i++) {
  const worker = new Worker('./counter-worker.js', {
    workerData: { sharedBuffer, increments: INCREMENTS_PER_WORKER }
  });

  worker.on('exit', () => {
    const final = Atomics.load(sharedArray, 0);
    console.log(`Final counter value: ${final}`);
  });
}

JAVASCRIPT

// counter-worker.js
const { parentPort, workerData } = require('worker_threads');

const { sharedBuffer, increments } = workerData;
const sharedArray = new Int32Array(sharedBuffer);

for (let i = 0; i < increments; i++) {
  Atomics.add(sharedArray, 0, 1);
}

parentPort.postMessage('done');

8. 通信方式の比較

通信方式	方向	特徴	適用シナリオ
`parentPort`	メインスレッド ↔ ワーカー	双方向メッセージング、構造化クローン	一般的なタスク間の通信
`workerData`	メインスレッド → ワーカー	読み取り専用；作成時に渡される	初期化設定／パラメータ
`MessageChannel`	ワーカー ↔ ワーカー	メインスレッドを経由せずにポート間で直接接続	スレッド間の連携
`SharedArrayBuffer`	すべてのスレッドで共有	ゼロコピー、Atomics が必要	高頻度のデータ交換

9. ワーカー vs child_process vs クラスター

機能	ワーカースレッド	child_process	クラスタ
ユニット	スレッド数	プロセス数	プロセス数
メモリ	共有プロセスメモリ	独立したメモリ空間	独立したメモリ空間
通信	メッセージ／共有メモリ	IPCのシリアライズ	IPCのシリアライズ
立ち上げコスト	中程度	高	高
適用対象	CPU負荷の高い計算	スタンドアロンプログラム／サンドボックス	マルチコアHTTPサービス
安定性	ワーカープロセスのクラッシュは、同じプロセスに影響を与える	子プロセスのクラッシュは互いに影響を与えない	ワーカープロセスのクラッシュは互いに影響を与えない
共有ステータス	SharedArrayBuffer	未対応	未対応

10. マルチスレッドに適したタスクと適さないタスク

マルチスレッド対応	マルチスレッド非対応
画像処理／サムネイル生成	簡単なHTTPリクエストのルーティング
暗号化・ハッシュ計算	データベースのCRUD操作
圧縮・解凍	ファイルI/O（非同期処理で十分）
大規模な数学演算	簡単なJSON変換
PDFの生成・レンダリング	短時間のマイクロタスク
オーディオおよびビデオのトランスコーディング	イベント駆動型メッセージ転送

11. スレッドプールパターン

ワーカーの作成にはある程度のオーバーヘッドが伴い、頻繁に作成・破棄を行うとリソースの無駄遣いになります。スレッドプールは一定数のワーカーを維持します。タスクが到着すると、それはアイドル状態のワーカーに割り当てられ、タスクが完了すると、そのワーカーは解放されて再利用されます。

▶ 例：シンプルなスレッドプールをゼロから構築する

JAVASCRIPT

// pool.js
const { Worker } = require('worker_threads');

class Pool {
  constructor(workerFile, size) {
    this.workerFile = workerFile;
    this.size = size;
    this.workers = [];
    this.queue = [];

    for (let i = 0; i < size; i++) {
      const worker = new Worker(workerFile);
      worker.busy = false;
      worker.on('message', (result) => {
        const task = worker.currentTask;
        worker.busy = false;
        worker.currentTask = null;
        task.resolve(result);
        this._processQueue();
      });
      worker.on('error', (err) => {
        const task = worker.currentTask;
        if (task) {
          worker.busy = false;
          worker.currentTask = null;
          task.reject(err);
          this._processQueue();
        }
      });
      this.workers.push(worker);
    }
  }

  run(data) {
    return new Promise((resolve, reject) => {
      const task = { data, resolve, reject };
      const idle = this.workers.find((w) => !w.busy);
      if (idle) {
        this._assign(idle, task);
      } else {
        this.queue.push(task);
      }
    });
  }

  _assign(worker, task) {
    worker.busy = true;
    worker.currentTask = task;
    worker.postMessage(task.data);
  }

  _processQueue() {
    if (this.queue.length === 0) return;
    const idle = this.workers.find((w) => !w.busy);
    if (!idle) return;
    const task = this.queue.shift();
    this._assign(idle, task);
  }

  destroy() {
    for (const worker of this.workers) {
      worker.terminate();
    }
    this.workers = [];
    this.queue = [];
  }
}

module.exports = Pool;

▶ 例：スレッドプールを使用したフィボナッチ数列の計算

JAVASCRIPT

// main.js
const Pool = require('./pool.js');

const pool = new Pool('./fib-worker.js', 4);

async function main() {
  const tasks = [40, 41, 42, 43, 44, 45, 46, 47];

  const promises = tasks.map((n) => pool.run({ n }));

  const results = await Promise.all(promises);

  for (let i = 0; i < tasks.length; i++) {
    console.log(`fib(${tasks[i]}) = ${results[i]}`);
  }

  pool.destroy();
}

main();

JAVASCRIPT

// fib-worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', ({ n }) => {
  const result = fib(n);
  parentPort.postMessage(result);
});

function fib(n) {
  if (n <= 1) return n;
  let a = 0, b = 1;
  for (let i = 2; i <= n; i++) {
    [a, b] = [b, a + b];
  }
  return b;
}

▶ 例：piscina スレッドプールライブラリ

BASH

npm install piscina

JAVASCRIPT

const path = require('path');
const Piscina = require('piscina');

const pool = new Piscina({
  filename: path.resolve(__dirname, 'task.js'),
  maxThreads: 4
});

async function main() {
  const results = await Promise.all([
    pool.run({ x: 10, y: 20 }),
    pool.run({ x: 30, y: 40 }),
    pool.run({ x: 50, y: 60 })
  ]);

  console.log(results); // [30, 70, 110]
  await pool.destroy();
}

main();

JAVASCRIPT

// task.js
module.exports = ({ x, y }) => {
  return x + y;
};

12. 総合的な例：画像処理のスレッドプール

これまでに学んだ知識を活用して、完全な画像処理スレッドプールシステムを構築してください。メインスレッドがタスクをディスパッチし、ワーカースレッドがサムネイルを処理し、その結果が収集・集計される仕組みです。

JAVASCRIPT

// image-pool.js
const { Worker } = require('worker_threads');
const path = require('path');

class ImagePool {
  constructor(workerCount) {
    this.workers = [];
    this.queue = [];
    this.results = [];

    for (let i = 0; i < workerCount; i++) {
      const worker = new Worker(path.join(__dirname, 'image-processor.js'));
      worker.busy = false;

      worker.on('message', (msg) => {
        if (msg.type === 'result') {
          this.results.push(msg.data);
        }
        worker.busy = false;
        worker.currentResolve();
        this._dispatch();
      });

      worker.on('error', (err) => {
        worker.busy = false;
        if (worker.currentReject) {
          worker.currentReject(err);
        }
        this._dispatch();
      });

      this.workers.push(worker);
    }
  }

  process(fileList) {
    this.results = [];
    const promises = fileList.map((file) => this._enqueue(file));
    return Promise.all(promises).then(() => this.results);
  }

  _enqueue(file) {
    return new Promise((resolve, reject) => {
      this.queue.push({ file, resolve, reject });
      this._dispatch();
    });
  }

  _dispatch() {
    while (this.queue.length > 0) {
      const idle = this.workers.find((w) => !w.busy);
      if (!idle) break;
      const task = this.queue.shift();
      idle.busy = true;
      idle.currentResolve = task.resolve;
      idle.currentReject = task.reject;
      idle.postMessage({ file: task.file, sizes: [200, 400, 800] });
    }
  }

  destroy() {
    this.workers.forEach((w) => w.terminate());
  }
}

module.exports = ImagePool;

JAVASCRIPT

// image-processor.js
const { parentPort } = require('worker_threads');

parentPort.on('message', async ({ file, sizes }) => {
  const results = [];

  for (const size of sizes) {
    const thumb = await resize(file, size);
    results.push(thumb);
  }

  parentPort.postMessage({
    type: 'result',
    data: { file, thumbnails: results }
  });
});

async function resize(file, maxSize) {
  return new Promise((resolve) => {
    const duration = Math.random() * 200 + 50;
    setTimeout(() => {
      resolve(`${file}_${maxSize}px.jpg`);
    }, duration);
  });
}

JAVASCRIPT

// run.js
const ImagePool = require('./image-pool.js');

async function main() {
  const pool = new ImagePool(4);

  const files = Array.from({ length: 20 }, (_, i) =>
    `photo-${String(i + 1).padStart(3, '0')}.jpg`
  );

  const start = Date.now();
  const results = await pool.process(files);
  const elapsed = Date.now() - start;

  console.log(`Processed ${files.length} images in ${elapsed} ms`);
  console.log(`Generated ${results.reduce((sum, r) => sum + r.thumbnails.length, 0)} thumbnails`);

  pool.destroy();
}

main();

BASH

node run.js

TEXT

Processed 20 images in 1234 ms
Generated 60 thumbnails

❓ よくある質問

Q ワーカースレッドとchild_processの違いは何ですか？

A ワーカースレッドは同じプロセス内で作成され、メモリ（SharedArrayBuffer）を共有できるため、通信オーバーヘッドが低くなります。一方、child_processはメモリが分離された個別のプロセスを作成し、シリアル化されたIPCを介して通信を行うため、オーバーヘッドは高くなりますが、セキュリティ面ではより安全です。

Q ワーカーはメインスレッドの変数にアクセスできますか？

A いいえ。ワーカーには独自の実行コンテキストがあり、メインスレッドの変数に直接アクセスすることはできません。メッセージのやり取りには postMessage を使用するか、初期化時に workerData を通じて値を渡すか、あるいは SharedArrayBuffer を使用してメモリを共有する必要があります。

Q ワーカースレッドはどのような場合に使用すべきですか？

A 画像処理、暗号化、圧縮・解凍、大規模な数学的計算、オーディオやビデオのトランスコーディングなど、CPU負荷の高いタスクの場合です。I/O負荷の高いタスクではワーカースレッドは必要ありません。Node.jsの非同期I/Oはすでに十分に効率的だからです。

Q SharedArrayBuffer は安全ですか？

A SharedArrayBuffer 自体には同期メカニズムは備わっていません。マルチスレッド環境での同時読み取りや書き込みは、レースコンディションを引き起こす可能性があります。原子性を確保するには、アトミック操作（add、load、store、compareExchange など）を使用するか、ロックを使用してアクセスを調整する必要があります。

Q ワーカーの作成には、大きなオーバーヘッドがかかりますか？

A ワーカーを作成するには、V8エンジンインスタンスとイベントループを初期化する必要があり、これには約30～50ミリ秒かかります。これはchild_processの場合よりも短いですが、それでも無視できるほどではありません。頻繁な作成や破棄を避けるため、スレッドプール内でワーカーを再利用することをお勧めします。

Q fs や http といった Node.js モジュールは、ワーカー内で使用できますか？

A はい。ワーカースレッドには完全な Node.js ランタイムが備わっており、組み込みモジュールの大部分をサポートしています。ただし、cluster などの一部のモジュールは、メインプロセスでのみ使用可能です。

Q 「Transferable」と「structured cloning」の違いは何ですか？

A structured cloning はデータをコピーするため、小規模なデータセットに適しています。一方、「Transferable」は所有権をターゲットスレッドに移譲します。これはゼロコピー操作ですが、元のスレッドはアクセス権を失います。大規模な ArrayBuffer などを扱うシナリオに適しています。

📖 まとめ

重要な概念とその活用方法
特集：画像処理の高速化に向けた「Alice」の基本概念と活用法
ワーカークラスの基本概念と使い方
parentPort：双方向通信の基本概念と活用方法
workerData 初期化データの基本概念と使用方法
MessageChannel および MessagePort の基本概念と使用方法
SharedArrayBuffer および Atomics の基本概念と使用方法
コミュニケーション手法の比較に関する主要な概念と活用法

📝 練習問題

このレッスンにあるすべてのコード例を完成させ、それぞれが正しく動作することを確認してください。
包括的な例を修正し、独自の拡張機能を追加する
公式ドキュメントを確認し、このレッスンで扱われていないAPIを1～2つ見つけ、それらのテストコードを作成してください。
振り返り：このレッスンで学んだことを、実際のプロジェクトにどのように活かしますか？
このレッスンで学んだことと、これまでのレッスンの内容を組み合わせて、小さなプロジェクトを作成してみてください。

Web-Tutorial 技術チーム

複数の開発者によって共同維持されているプログラミングチュートリアルプラットフォーム。各チュートリアルは専門分野の開発者が執筆・レビューしています。正確で信頼性の高いコンテンツを目指しています — 問題を見つけた場合はお知らせください。

私たちについて

404 Not Found

ワーカースレッド

1. 学習内容

2. ストーリー：アリスの画像処理の高速化

3. ワーカークラスの基礎

▶ 例：基本的なワーカーの作成と通信

4. parentPort 双方向通信

▶ 例：双方向のメッセージ交換

▶ 例：転送可能なオブジェクトのゼロコピー転送

5. workerData の初期化データ

▶ 例：画像バッチ処理ワーカー

6. MessageChannel と MessagePort

▶ 例：2人の作業員が直接やり取りする場合

7. SharedArrayBuffer とアトミック操作

▶ 例：共有カウンター

8. 通信方式の比較

9. ワーカー vs child_process vs クラスター

10. マルチスレッドに適したタスクと適さないタスク

11. スレッドプールパターン

▶ 例：シンプルなスレッドプールをゼロから構築する

▶ 例：スレッドプールを使用したフィボナッチ数列の計算

▶ 例：piscina スレッドプールライブラリ

12. 総合的な例：画像処理のスレッドプール

❓ よくある質問

📖 まとめ

📝 練習問題