Posted inDevelopment

Node.jsにおけるBullMQとRedisを使ったJob Queue管理実践ガイド：Concurrency制御、Delayed Jobs、自動Retry

深夜2時のサーバー過負荷トラブル

午前2時ちょうど、PagerDutyのアラート通知が鳴り響きました。Datadogのダッシュボードを見ると、API Gatewayが真っ赤な504 Gateway Timeoutエラーを返しています。4台のEC2インスタンスのCPU使用率は100%に張り付き、RAMは枯渇、Node.jsプロセスはOut of Memory（OOM）でクラッシュを繰り返していました。

原因は深夜に実行されたメールマーケティングのcronjobでした。システムがリクエストを受け取り、Expressのリクエストハンドラー内で直接処理を実行していたのです。バックエンドはHTMLテンプレートのレンダリング、2MBの請求書PDFの圧縮、そして20,000人以上の顧客に対する一斉メール送信API呼び出しをすべて同時に処理していました。

リクエスト・レスポンスサイクル内に重い処理を詰め込むのは典型的なアンチパターンです。たとえPromise.all()でラップしたとしても、Node.jsはシングルスレッドのEvent Loop上で動作しています。CPU集約型のタスクや外部I/O待ちが数百件同時に発生すれば、Event Loopは完全にブロックされます。その結果、最も軽量な/healthzエンドポイントですらタイムアウトに巻き込まれてしまいます。この問題を根本的に解決するには、RedisとBullMQによるJob Queueアーキテクチャを導入し、リクエストの受付とバックグラウンド処理を完全に分離する必要があります。

Job Queueの仕組みと基本モデル

キューアーキテクチャは、おなじみのProducer – Consumerモデルに基づいて動作します：

  • Producer (Web API): クライアントからリクエストを受け取り、ジョブデータ（job payload）をパッケージ化してRedisに投入します。その後すぐにHTTP 202 Acceptedをユーザーに返します。この処理はユーザーを10〜30秒待たせることなく、わずか2〜5ms程度で完了します。
  • Queue Broker (Redis): サブミリ秒単位の読み書き速度を誇るRAM上のキューバッファとして機能します。
  • Worker / Consumer: Webサーバーとは独立してバックグラウンドで動作する別個のNode.jsプロセスです。Redisからジョブを順次または割り当てられたリソース制限に従って並行してフェッチし、処理を実行します。

BullMQは、現在Node.js向けとして最も支持されているキュー管理ライブラリであり、完全にTypeScriptで記述されています。従来のBullがRedisのPub/Subやポーリングに依存していたのに対し、BullMQはRedis StreamsとアトミックなLuaスクリプトを最大限に活用しています。これにより、タスクを正確に分散し、concurrency制御、delayed jobs、自動リトライを安全にサポートし、Workerが突然シャットダウンした場合でもデータ損失の心配がありません。

BullMQとRedisの段階的セットアップ

1. 環境の準備

Dockerを使ってRedis 7のコンテナをすばやく立ち上げます：

docker run -d --name redis-queue -p 6379:6379 redis:7-alpine

プロジェクトに必要な2つのコアライブラリをインストールします：

npm install bullmq ioredis

2. 安全なRedis接続設定

接続を再利用するためにredis.jsファイルを作成します。ここで極めて重要な注意点として、maxRetriesPerRequest: nullの設定が必須となります。BullMQは新規ジョブを待機するためにBRPOPLPUSHやBLMOVEなどのブロッキングコマンドを使用するため、ioredisのデフォルトのリトライ上限が有効になっていると、途中で接続が切断されてしまいます。

// redis.js
const IORedis = require('ioredis');

const redisConnection = new IORedis({
  host: process.env.REDIS_HOST || '127.0.0.1',
  port: Number(process.env.REDIS_PORT) || 6379,
  maxRetriesPerRequest: null,
  enableReadyCheck: false,
});

module.exports = redisConnection;

3. Queueの定義とジョブの投入

queue.jsでキューを初期化し、履歴データの保持ポリシーおよびリトライ戦略（Retry Strategy）を設定します：

// queue.js
const { Queue } = require('bullmq');
const redisConnection = require('./redis');

const emailQueue = new Queue('email-processing', {
  connection: redisConnection,
  defaultJobOptions: {
    // 確認用に完了した直近1,000件のジョブを保持
    removeOnComplete: {
      count: 1000,
    },
    // 後でエンジニアがログ調査できるようエラーとなった5,000件を保持
    removeOnFail: {
      count: 5000,
    },
    // エラー発生時は最大5回までリトライ
    attempts: 5,
    // 指数バックオフで再試行間隔を延長: 3秒、6秒、12秒、24秒...
    backoff: {
      type: 'exponential',
      delay: 3000,
    },
  },
});

async function addEmailJob(data) {
  return await emailQueue.add('send-welcome-email', data);
}

module.exports = { emailQueue, addEmailJob };

4. Delayed Jobs（遅延ジョブ）の設定

毎分cronjobを実行してデータベースのordersテーブルをポーリングする代わりに、delayパラメータを使ってジョブの実行時刻を直接スケジューリングできます。単位はミリ秒です：

// カゴ落ちリマインドメールを15分後（900,000ミリ秒）に送信
await emailQueue.add(
  'send-cart-reminder',
  { userId: 'user_102', cartId: 'cart_883' },
  { delay: 15 * 60 * 1000 }
);

5. Workerの構築とConcurrency制御

Workerは実際にCPUとRAMを消費する場所です。サーバーのリソースが2 vCPU程度しかない場合、concurrencyを50や100に設定すべきではありません。この数値を適正な範囲（5〜10程度）に抑えることで、CPUリソースを保護し、PostgreSQLやMySQLのコネクションプール枯渇を防ぐことができます。

// worker.js
const { Worker } = require('bullmq');
const redisConnection = require('./redis');

// 30%の確率でエラーが発生するサードパーティメール送信（SendGrid、SESなど）のモック
async function mockSendEmail(jobData) {
  if (Math.random() < 0.3) {
    throw new Error('503 Service Unavailable: メールプロバイダのゲートウェイがタイムアウトしました！');
  }
  console.log(`[成功] メールを正常に送信しました: ${jobData.email}`);
}

const emailWorker = new Worker(
  'email-processing',
  async (job) => {
    console.log(`[処理中] ジョブ ${job.id} - 試行 ${job.attemptsMade + 1} 回目`);
    await mockSendEmail(job.data);
  },
  {
    connection: redisConnection,
    concurrency: 5, // このプロセスで最大5ジョブを並行処理
  }
);

emailWorker.on('completed', (job) => {
  console.log(`[完了] ジョブ ${job.id} が正常に終了しました。`);
});

emailWorker.on('failed', (job, err) => {
  console.error(`[失敗] ジョブ ${job.id} が試行 ${job.attemptsMade} 回目で失敗しました: ${err.message}`);
});

6. 自動復帰メカニズムの検証

server.jsを作成し、20件のジョブをキューに一括投入してシステムの挙動を観察します：

// server.js
const { addEmailJob } = require('./queue');

async function bootstrap() {
  console.log('キューへの20件のジョブ投入を開始します...');
  for (let i = 1; i <= 20; i++) {
    await addEmailJob({
      email: `customer_${i}@example.com`,
      subject: 'アカウントの有効化',
    });
  }
  console.log('Producerは数ミリ秒で20件すべてのジョブ投入を完了しました。');
}

bootstrap();

1つ目のターミナルを開き、Workerを起動します：

node worker.js

2つ目のターミナルを開き、負荷を投入します：

node server.js

Workerターミナルの出力結果から、キューによる保護機構が確認できます：

  1. システムは同時に最大5件のジョブのみをフェッチして処理し、CPU使用率を15〜25%前後の安定した状態に保ちます。
  2. モックのタイムアウトエラーに遭遇したジョブも破棄されません。Exponential Backoffに従って待機間隔を徐々に伸ばしながら、自動的にリトライキューに再配置されます。
  3. 外部メールサービスが復旧すると、エンジニアの手動介入を必要とせずにすべての失敗ジョブが正常に完了します。

本番環境における実践的な運用の勘所

ProducerとWorkerを分離することで負荷を即座に軽減できますが、状態管理の複雑さは増します。

最も重要なのは冪等性（Idempotency）の原則です。Workerが処理の途中でクラッシュした場合や、メール送信APIがネットワーク層でタイムアウトした場合、BullMQはそのジョブを再実行します。送信前に状態チェックを行わないコードになっていると、顧客に同一のメールが3〜4通届いてしまったり、さらに深刻な場合は二重決済が発生したりする恐れがあります。主要なタスクを実行する前に、必ずデータベース内のステータスフラグを確認するか、Idempotency Key（冪等キー）を使用してください。

次に考慮すべきはRedisのRAM容量です。ジョブのメタデータとペイロードは、平均して1件あたり1KB〜4KBのメモリを消費します。毎日200,000件のジョブを処理するシステムでremoveOnCompleteの設定を忘れると、履歴データの蓄積だけでRedisは毎日約1GBのRAMを消費し続けます。そのまま放置すれば、2〜3週間でRedisクラスタのメモリが枯渇しダウンします。

最後に、WorkerとAPIサーバーは必ず別々のコンテナやインスタンスに分離してデプロイしてください。Workerが動画のレンダリングや重いPDFの圧縮処理でCPUを使い果たしたとしても、APIサーバーは影響を受けず、ユーザーからのリクエストを遅延なくスムーズに処理し続けることができます。

まとめ

負荷の高い処理をメインのリクエストフローから切り離すことは、504タイムアウトエラーを根絶し、Node.jsの高い応答性を維持するための基盤となるアプローチです。BullMQとRedisを組み合わせることで、concurrencyによるシステム負荷の制御、外部サービス不安定時の自動リトライ、無駄なcronjobに頼らない柔軟なジョブスケジューリングを確実に実現できます。

Share:
Tags: