BullMQ、Redis、TypeScriptを用いた分散タスクキューの構築

Node.jsにおけるバックグラウンド処理手法の比較

APIエンドポイントが重い処理（50MBのPDF生成、4K動画のトランスコード、外部Webhookへの数千件のデータ同期など）を処理すると、HTTPレスポンス時間は急速に悪化します。リクエスト/レスポンスのライフサイクルからこれらのタスクを切り離す（オフロードする）ことが極めて重要です。本番環境のNode.jsシステムでは、通常以下の3つの主要パターンから選択します：

• インメモリのマイクロキュー（例: fastq, EventEmitter）: 外部依存関係がなく最も迅速に実装できます。しかし、デプロイ等でNode.jsプロセスが再起動すると、メモリ上のキューに入っていたジョブはすべて消失します。また、複数のサーバーインスタンス間でワークロードを分散することもできません。
• 重量級エンタープライズブローカー（例: Apache Kafka, RabbitMQ）: 毎秒数百万件のイベントのストリーミングや、複雑なメッセージトポロジーの管理に優れています。デメリットは運用のオーバーヘッドが大きく、学習コストが高いこと、また単純なタスクスケジューリングや遅延リトライに対してセットアップが複雑になる点です。
• Redisベースの分散キュー（BullMQ）: 近年のTypeScriptおよびNode.jsマイクロサービスに最適な選択肢です。BullMQはRedis Streams、Sorted Sets、アトミックなLuaスクリプトを活用し、重量級のインフラを抱えることなく、分散ワーカー間でのスケジューリング、リトライ、レート制限を効率的に処理します。

BullMQ + Redisアーキテクチャのメリットとデメリット

BullMQを標準採用する前に、そのアーキテクチャが自社のインフラ要件とどのように合致するかを評価してください：

メリット

• ネイティブな遅延＆Cronジョブ: 外部のcronデーモンを保守することなく、15分後にタスクを実行したり、毎晩UTC午前0時に定期実行したりできます。
• アトミックな状態遷移: BullMQはRedis内で直接Luaスクリプトを実行します。これにより、競合状態（レースコンディション）を起こすことなく、waitingからactive、completed、failedへのアトミックなステータス更新が保証されます。
• 厳格なTypeScriptの型安全性: ジョブデータと戻り値の型にジェネリクスを使用することで、プロデューサーとコンシューマー間のスキーマの不整合をコンパイル時に検知できます。
• きめ細かな並行処理とレート制限: ワーカーの並行実行数を制限し、全体のスループットを制御（例: 10個のPod全体で毎秒50リクエストに制限）することで、サードパーティAPIのレート制限超過を防ぐことができます。

デメリットとトレードオフ

• Redisのメモリフットプリント: ジョブのペイロードと実行メタデータはRAM上に保持されます。厳格な保持ポリシーを設定せずに1日50万件のジョブを処理すると、数日でRedisのメモリが枯渇する可能性があります。
• 専用コネクションのオーバーヘッド: BullMQのワーカーはブロッキングRedisコマンドを使用してジョブを即座に取得します。各キューとワーカーインスタンスは複数の専用TCP接続を維持します。

推奨アーキテクチャと本番環境のセットアップ

信頼性の高いタスク処理システムは、処理を2つの明確な役割に分離します。タスクを発行するプロデューサー（APIサーバー）と、タスクを実行するコンシューマー（専用のワーカープロセス）です。両者は共有のRedisクラスターを介して通信します。

本番パイプラインにおける基本パターンは以下のとおりです：

1. プライマリキュー: 指数バックオフ（例: 初回2秒の遅延で試行ごとに倍増）が設定された標準ジョブを受け取ります。
2. デッドレターキュー（DLQ）: すべてのリトライ試行に失敗したペイロードをキャプチャし、通常トラフィックをブロックすることなく、エンジニアがスタックトレースを調査してジョブを再実行できるようにします。
3. 接続の分離: 長時間のブロッキングコマンドを許可するため、BullMQの要件に従ってioredisにmaxRetriesPerRequest: nullを設定します。

実装ガイド

1. プロジェクトのセットアップと接続設定

TypeScriptプロジェクトを初期化し、必要な依存パッケージをインストールします：

npm install bullmq ioredis dotenv
npm install -D typescript @types/node tsx

src/config/redis.tsで共有Redis接続を設定します：

import { ConnectionOptions } from 'bullmq';

export const redisConnection: ConnectionOptions = {
  host: process.env.REDIS_HOST || '127.0.0.1',
  port: Number(process.env.REDIS_PORT) || 6379,
  password: process.env.REDIS_PASSWORD || undefined,
  maxRetriesPerRequest: null,
  enableReadyCheck: false,
};

2. ジョブ型とプロデューサーの定義

契約の安全性を担保するために、src/types/queue.tsで明確なペイロードインターフェースを定義します：

export interface EmailJobData {
  to: string;
  subject: string;
  templateId: string;
  variables: Record<string, string>;
}

export interface EmailJobResult {
  messageId: string;
  sentAt: string;
}

export const EMAIL_QUEUE_NAME = 'email-delivery';
export const EMAIL_DLQ_NAME = 'email-delivery-dlq';

src/queues/emailQueue.tsでキュープロデューサーをインスタンス化します：

import { Queue } from 'bullmq';
import { redisConnection } from '../config/redis';
import { EMAIL_QUEUE_NAME, EmailJobData, EmailJobResult } from '../types/queue';

export const emailQueue = new Queue<EmailJobData, EmailJobResult>(
  EMAIL_QUEUE_NAME,
  {
    connection: redisConnection,
    defaultJobOptions: {
      attempts: 5,
      backoff: {
        type: 'exponential',
        delay: 2000, // 2秒、4秒、8秒、16秒、32秒の間隔でリトライ
      },
      removeOnComplete: {
        age: 3600, // 完了したジョブのメタデータを1時間保持
        count: 1000,
      },
      removeOnFail: false, // 調査用に失敗したジョブを保持
    },
  }
);

export async function scheduleEmail(data: EmailJobData, delayMs: number = 0) {
  const job = await emailQueue.add('send-transactional-email', data, {
    delay: delayMs,
  });
  return job.id;
}

3. デッドレタールーティングを備えたワーカーの構築

外部APIで一時的なネットワークエラーが発生した場合、BullMQは指数バックオフを用いて自動的にタスクをリトライします。5回の試行がすべて失敗した場合は、エラーのコンテキストをキャプチャし、調査およびアラート通知のためにデッドレターキューに直接送信します。

src/workers/emailWorker.tsでコンシューマーワーカーを作成します：

import { Worker, Queue, Job } from 'bullmq';
import { redisConnection } from '../config/redis';
import {
  EMAIL_QUEUE_NAME,
  EMAIL_DLQ_NAME,
  EmailJobData,
  EmailJobResult,
} from '../types/queue';

const deadLetterQueue = new Queue(EMAIL_DLQ_NAME, { connection: redisConnection });

export const emailWorker = new Worker<EmailJobData, EmailJobResult>(
  EMAIL_QUEUE_NAME,
  async (job: Job<EmailJobData, EmailJobResult>) => {
    // ダウンストリームAPIの障害をシミュレート
    if (job.data.to.includes('fail')) {
      throw new Error(`${job.data.to} に対するアップストリームSMTPプロバイダーのタイムアウトが発生しました`);
    }

    return {
      messageId: `msg_${job.id}_${Date.now()}`,
      sentAt: new Date().toISOString(),
    };
  },
  {
    connection: redisConnection,
    concurrency: 10, // ワーカーインスタンスあたり10ジョブを並行処理
    limiter: {
      max: 100,
      duration: 1000, // ワーカーあたりのスループットを毎秒100ジョブに制限
    },
  }
);

emailWorker.on('completed', (job: Job) => {
  console.log(`[Job ${job.id}] ${job.data.to} への送信が正常に完了しました`);
});

emailWorker.on('failed', async (job: Job | undefined, error: Error) => {
  if (!job) return;

  console.warn(`[Job ${job.id}] 試行回数 ${job.attemptsMade}/${job.opts.attempts} で失敗: ${error.message}`);

  // すべての試行が失敗した場合、DLQにルーティング
  if (job.attemptsMade >= (job.opts.attempts || 1)) {
    console.error(`[Job ${job.id}] リトライ上限に達しました。DLQにルーティングします: ${EMAIL_DLQ_NAME}`);
    await deadLetterQueue.add('failed-email-poison-pill', {
      originalJobId: job.id,
      originalData: job.data,
      failedReason: error.message,
      failedAt: new Date().toISOString(),
      stacktrace: error.stack,
    });
  }
});

4. グレースフルシャットダウンとシグナル処理

タスクの処理中にワーカーを突然停止させると、ジョブが孤立したりデータが破損したりするリスクがあります。実行ファイル（src/index.ts）でシステムの終了シグナル（SIGTERM、SIGINT）を必ずインターセプトし、アクティブなジョブが完了できるようにしてください：

import { emailWorker } from './workers/emailWorker';
import { emailQueue } from './queues/emailQueue';

async function shutdown(signal: string) {
  console.log(`${signal} を受信しました。実行中のタスクをドレインし、接続をクローズしています...`);
  
  // 実行中のジョブを完了させ、新しいタスクの受付を停止
  await emailWorker.close();
  await emailQueue.close();
  
  console.log('タスクキューのシャットダウンが完了しました。正常終了します。');
  process.exit(0);
}

process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));

高トラフィックな本番システム向けのベストプラクティス

• ペイロードを軽量に保つ: 10MBのファイルバッファをジョブデータに直接埋め込むのではなく、IDやオブジェクトストレージのURL（例: S3のパス）を渡すようにします。
• ハンドラーの冪等性を担保する: 一時的なネットワークエラーによりリトライが発生するため、同じジョブが複数回実行される可能性があります。データベースのトランザクションIDやUPSERT処理を活用し、重複請求やメールの多重送信を防いでください。
• Redisメモリのトリミングを自動化する: 必ずremoveOnCompleteおよびremoveOnFailの上限を設定してください。また、Redisのメモリ使用率が75%を超えた場合にアラートを発報するよう監視を設定します。
• キューの滞留数とレイテンシーを監視する: PrometheusやDatadogなどのツールでqueue.getJobCounts()を用いて待機中および実行中のジョブ数を追跡します。待機時間が5秒を超える場合は、ワーカーPodを動的にスケールアウトさせます。