0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@blinkgroup_jp(ブリンクグループ)

Node.jsで重い処理を分離する：Worker Threadsとメッセージキュー（BullMQ）の選定基準と実装パターン

0
Posted at

Node.jsはシングルスレッド（イベントループ）で動作するため、CPU負荷の高い処理（暗号化、画像・動画解析、大規模なデータ集計など）を実行すると、イベントループがブロックされ、APIのレスポンス遅延やシステム全体のパフォーマンス低下を引き起こします。

この記事では、重い処理をメインプロセスから分離するための2つの主要なアプローチである「Worker Threads」と「メッセージキュー（BullMQ）」について、具体的な選定基準、実装コード例、および実務での注意点を解説します。

この記事で分かること

  • Worker Threadsとメッセージキューの使い分け基準（比較表）
  • Worker Threadsを用いたCPU負荷処理の分離実装
  • BullMQ（Redis）を用いた非同期ジョブキューの実装
  • 実務導入時の注意点とトラブルシューティング

対象読者・前提条件

  • Node.js（Express, NestJSなど）でバックエンドAPIを開発しているエンジニア
  • サービスのスケールに伴い、重い処理のバックグラウンド化を検討している方
  • Node.js v18以降、Redisが利用可能な環境を想定しています。

Worker Threads と メッセージキューの比較

どちらの手法を採用すべきかは、処理の性質やインフラ構成によって異なります。

比較項目 Worker Threads メッセージキュー (例: BullMQ + Redis)
主な目的 同一インスタンス内でのマルチスレッド化 プロセス・サーバー間での非同期分散処理
適した処理 短時間で終わるCPU集中的な計算（画像リサイズ、暗号化など） 時間がかかる処理、外部API連携、リトライが必要なバッチ処理
リソース共有 メモリ（ArrayBufferなど）を共有可能 メモリ共有不可（シリアライズされたデータ転送）
スケーラビリティ 単一サーバーのCPUコア数に依存 複数サーバー（Workerプロセス）への水平分散が可能
耐障害性 プロセスがクラッシュするとWorkerも停止 ジョブがRedisに永続化され、再起動後にリトライ可能
導入コスト 低（外部ミドルウェア不要） 中〜高（Redisの構築・運用が必要）

パターン1：Worker ThreadsによるCPU負荷処理の分離

同一サーバー内でCPUリソースを効率的に活用したい場合、標準モジュールの worker_threads を使用します。

実装例：重い計算処理（素数判定など）の分離

以下は、メインプロセスからWorkerを起動し、処理を委譲する実装例です。

1. worker.js（ワーカースレッド側）

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

// 重い計算処理の例
function heavyComputation(limit) {
  let count = 0;
  for (let i = 2; i <= limit; i++) {
    let isPrime = true;
    for (let j = 2; j <= Math.sqrt(i); j++) {
      if (i % j === 0) {
        isPrime = false;
        break;
      }
    }
    if (isPrime) count++;
  }
  return count;
}

// メインプロセスからのデータを受け取り処理を実行
const result = heavyComputation(workerData.limit);

// 結果をメインプロセスに送信
parentPort.postMessage({ success: true, result });

2. main.js（メインプロセス側）

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

function runWorker(limit) {
  return new Promise((resolve, reject) => {
    const worker = new Worker(path.resolve(__dirname, 'worker.js'), {
      workerData: { limit }
    });

    worker.on('message', (message) => {
      resolve(message.result);
    });

    worker.on('error', (error) => {
      reject(error);
    });

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

// Expressのハンドラーなどでの利用イメージ
async function handleRequest(req, res) {
  try {
    const limit = parseInt(req.query.limit, 10) || 1000000;
    const result = await runWorker(limit);
    res.json({ result });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
}

パターン2：BullMQによる非同期ジョブキュー処理

処理時間が数秒〜数分以上に及ぶ場合や、リトライ制御、レートリミット、複数サーバーへの負荷分散が必要な場合は、Redisをバックエンドとした BullMQ などのメッセージキューを採用します。

実装例：非同期タスクのキュー登録と処理

事前に npm install bullmq ioredis でライブラリをインストールし、Redisサーバーを起動しておいてください。

1. queue.js（ジョブの登録側）

const { Queue } = require('bullmq');
const IORedis = require('ioredis');

const connection = new IORedis({ host: 'localhost', port: 6379 });
const reportQueue = new Queue('report-generation', { connection });

async function addReportJob(userId, reportType) {
  // ジョブをキューに追加（即座にレスポンスを返す）
  const job = await reportQueue.add('generate', {
    userId,
    reportType,
    createdAt: new Date().toISOString()
  }, {
    attempts: 3, // 失敗時のリトライ回数
    backoff: {
      type: 'exponential',
      delay: 5000 // リトライ間隔（5秒から指数関数的に増加）
    }
  });
  return job.id;
}

module.exports = { addReportJob };

2. worker-process.js（ジョブの処理側 / 別プロセス）

const { Worker } = require('bullmq');
const IORedis = require('ioredis');

const connection = new IORedis({ host: 'localhost', port: 6379 });

const worker = new Worker('report-generation', async (job) => {
  console.log(`Job ${job.id} processing for user ${job.data.userId}...`);
  
  // 実際の重い処理（PDF生成や外部API連携など）をシミュレート
  await new Promise((resolve) => setTimeout(resolve, 10000));

  console.log(`Job ${job.id} completed successfully.`);
  return { downloadUrl: `https://example.com/reports/${job.id}.pdf` };
}, { connection });

worker.on('failed', (job, err) => {
  console.error(`Job ${job.id} failed with error: ${err.message}`);
});

導入時の注意点とトラブルシューティング

Worker Threadsの注意点

  1. スレッド生成のオーバーヘッド
    リクエストのたびに new Worker() を実行すると、スレッド起動のオーバーヘッドにより逆にパフォーマンスが低下します。実務では、スレッドを再利用する「スレッドプール（例: piscina ライブラリなど）」の導入を検討してください。
  2. I/O処理の混在
    Worker ThreadsはCPU負荷処理に特化しています。データベースアクセスや外部API呼び出しなどのI/O待ちが多い処理は、Worker Threadsを使ってもパフォーマンス向上は限定的です。非同期I/O（async/await）を適切に使用するか、メッセージキューに逃がす方が適しています。

メッセージキュー（BullMQ）の注意点

  1. データのシリアライズ制限
    キューに渡す引数（job.data）は、Redisに保存するためにJSON文字列化されます。クラスのインスタンスや関数、循環参照を含むオブジェクトは直接渡せません。必要なIDやプリミティブな値のみを渡すように設計してください。
  2. ジョブの重複実行への考慮
    ネットワークの一時的な瞬断やプロセスの再起動により、同一のジョブが複数回実行される可能性があります。処理対象のステータスチェックを行うなど、冪等性（べきとうせい）を担保した実装にしてください。

まとめと選定チェックリスト

どちらの手法を採用すべきか迷った際は、以下のチェックリストを基準に判断してください。

  • 処理時間は1秒未満か？
    • Yes → Worker Threads（同一サーバー内で高速に処理を完了させる）
    • No → メッセージキュー（ユーザーを待たせずにバックグラウンドで非同期処理）
  • 処理が失敗した際に自動リトライさせたいか？
    • Yes → メッセージキュー（指数バックオフなどの高度なリトライ制御が可能）
  • インフラ構成をシンプルに保ちたいか（外部ミドルウェアを増やしたくないか）？
    • Yes → Worker Threads（Node.js標準機能のみで完結）
    • No（Redisの導入・運用が可能） → メッセージキュー
  • 将来的に処理専用のサーバーをスケールアウト（水平分散）させたいか？
    • Yes → メッセージキュー（Workerプロセスを別サーバーに分離可能）

システムの要件と運用コストのバランスを考慮し、最適な分離手法を選択してください。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?