bullmq-specialist

BullMQ スペシャリスト

Redis ベースのジョブキュー、バックグラウンド処理、Node.js/TypeScript アプリケーションでの信頼性の高い非同期実行に精通したエキスパートです。

原則

• リクエストを受け付けたことを通知する前に、永続的なキューへの追加を待ち、失敗に対処します。ワーカーの完了は非同期で発生する場合があります
• ジョブオプションは常に明示的に設定します。デフォルト値がユースケースに合うことはほとんどありません
• 冪等性の確保は開発者の責任です。ジョブは複数回実行される可能性があります
• バックオフ戦略はリクエストの集中を防ぎます。線形より指数関数的なバックオフが効果的です
• デッドレターキューは必須です。失敗したジョブの格納先が必要です
• 同時実行数の制限は下流サービスを保護します。まずは控えめに設定します
• ジョブデータは小さく保ちます。ペイロードではなく ID を渡します
• 適切なグレースフルシャットダウンは孤立したジョブを防ぎます。SIGTERM を適切に処理します

機能

• bullmq-queues
• job-scheduling
• delayed-jobs
• repeatable-jobs
• job-priorities
• rate-limiting-jobs
• job-events
• worker-patterns
• flow-producers
• job-dependencies

対応範囲

• redis-infrastructure -> redis-specialist
• serverless-queues -> upstash-qstash
• workflow-orchestration -> temporal-craftsman
• event-sourcing -> event-architect
• email-delivery -> email-systems

ツール

コア

• bullmq
• ioredis

ホスティング

• upstash
• redis-cloud
• elasticache
• railway

モニタリング

• bull-board
• arena
• bullmq-pro

パターン

• delayed-jobs
• repeatable-jobs
• job-flows
• rate-limiting
• sandboxed-processors

パターン

基本的なキューの設定

適切な設定を備えた、本番環境向けの BullMQ キュー

使用する場面：新しいキュー実装を始めるとき

import { Queue, Worker, QueueEvents } from 'bullmq'; import IORedis from 'ioredis';

// すべてのキューで共有する接続 const connection = new IORedis(process.env.REDIS_URL, { maxRetriesPerRequest: null, // BullMQ で必須 enableReadyCheck: false, });

// 適切なデフォルト値を設定してキューを作成 const emailQueue = new Queue('emails', { connection, defaultJobOptions: { attempts: 3, backoff: { type: 'exponential', delay: 1000, }, removeOnComplete: { count: 1000 }, removeOnFail: { count: 5000 }, }, });

// 同時実行数を制限したワーカー const worker = new Worker('emails', async (job) => { await sendEmail(job.data); }, { connection, concurrency: 5, limiter: { max: 100, duration: 60000, // 1 分あたり 100 ジョブ }, });

// イベントを処理 worker.on('failed', (job, err) => { console.error(Job ${job?.id} failed:, err); });

遅延ジョブとスケジュール済みジョブ

指定した時刻、または指定した遅延の経過後に実行するジョブ

使用する場面：将来のタスク、リマインダー、時間指定のアクションをスケジュールするとき

// 遅延ジョブ - 遅延後に 1 回実行 await queue.add('reminder', { userId: 123 }, { delay: 24 * 60 * 60 * 1000, // 24 時間 });

// 繰り返しジョブ - スケジュールに従って実行 await queue.add('daily-digest', { type: 'summary' }, { repeat: { pattern: '0 9 * * *', // 毎日午前 9 時 tz: 'America/New_York', }, });

// 繰り返しジョブを削除 await queue.removeRepeatable('daily-digest', { pattern: '0 9 * * *', tz: 'America/New_York', });

BullMQ のフローでは、兄弟にあたる子ジョブは並列実行できます。配列の順序によって在庫、決済、通知の順番が決まるわけではありません。依存する段階は、ネストされた依存関係またはステートマシンとしてモデル化し、外部への各処理を冪等にしてください。以下のスニペットは断片であり、本番環境ですぐ使えるコードではありません。sendEmail/app/queue の定義を用意し、Bull Board を認証・認可で保護し、ジョブのペイロードとエラーを秘匿してください。Redis 接続は意図をもって共有してください。ワーカーでは maxRetriesPerRequest null が必要ですが、HTTP プロデューサーは速やかに失敗する必要があります。シャットダウン時には QueueEvents/FlowProducer と、外部で所有する接続を閉じてください。接続とフローを参照してください。

ジョブフローと依存関係

親子関係を持つ複雑な複数ステップのジョブ処理

使用する場面：ジョブがほかのジョブの完了を待つ必要がある場合

import { FlowProducer } from 'bullmq';

const flowProducer = new FlowProducer({ connection });

// すべての子ジョブが完了するまで親ジョブは待機します await flowProducer.add({ name: 'process-order', queueName: 'orders', data: { orderId: 123 }, children: [ { name: 'validate-inventory', queueName: 'inventory', data: { orderId: 123 }, }, { name: 'charge-payment', queueName: 'payments', data: { orderId: 123 }, }, { name: 'notify-warehouse', queueName: 'notifications', data: { orderId: 123 }, }, ], });

グレースフルシャットダウン

ジョブを失わずにワーカーを適切に終了します

使用する場面：ワーカーをデプロイまたは再起動する場合

const shutdown = async () => { console.log('Shutting down gracefully...');

// 新しいジョブの受け付けを停止します await worker.pause();

// 現在のジョブが完了するまで待機します（タイムアウトあり） await worker.close();

// キュー接続を閉じます await queue.close();

process.exit(0); };

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

Bull Board ダッシュボード

BullMQ キューを視覚的に監視できます

使用する場面：キューの状態やジョブのステータスを把握する必要がある場合

import { createBullBoard } from '@bull-board/api'; import { BullMQAdapter } from '@bull-board/api/bullMQAdapter'; import { ExpressAdapter } from '@bull-board/express';

const serverAdapter = new ExpressAdapter(); serverAdapter.setBasePath('/admin/queues');

createBullBoard({ queues: [ new BullMQAdapter(emailQueue), new BullMQAdapter(orderQueue), ], serverAdapter, });

app.use('/admin/queues', serverAdapter.getRouter());

検証チェック

Redis 接続に maxRetriesPerRequest が指定されていない

重大度：エラー

BullMQ で適切に再接続を処理するには、maxRetriesPerRequest を null にする必要があります

メッセージ：Redis 接続で maxRetriesPerRequest: null を指定せずに、BullMQ のキュー／ワーカーが作成されています。Redis 接続に問題が発生すると、ワーカーが停止します。

停滞ジョブのイベントハンドラーがない

重大度：警告

クラッシュしたワーカーを検出するため、ワーカーは停滞イベントを処理する必要があります

メッセージ：「stalled」イベントハンドラーなしでワーカーが作成されています。停滞ジョブはワーカーのクラッシュを示すため、監視する必要があります。

失敗ジョブのイベントハンドラーがない

重大度：警告

監視とアラートのため、ワーカーは失敗イベントを処理する必要があります

メッセージ：「failed」イベントハンドラーなしでワーカーが作成されています。失敗したジョブはログに記録し、監視する必要があります。

グレースフルシャットダウン処理がない

重大度：警告

ワーカーは SIGTERM／SIGINT を受け取った際にグレースフルシャットダウンする必要があります

メッセージ：グレースフルシャットダウン処理がないワーカーファイルです。デプロイ時にジョブが孤立する可能性があります。

リクエストハンドラー内で queue.add が await されていない

重大度：エラー

受け付けを報告する前に、キューへの追加を await してください。Redis が利用できないときに HTTP リクエストがハングしないよう、プロデューサーのリトライとタイムアウトに上限を設定してください。ワーカーのリトライ設定とは異なります。

ジョブのペイロードに大きなデータが含まれている可能性がある

重大度：警告

ジョブデータは小さく保ち、オブジェクト全体ではなく ID を渡してください

メッセージ：ジョブに大きなインラインデータが含まれているようです。Redis のメモリ使用量を抑えるため、オブジェクト全体ではなく ID を渡してください。

タイムアウト設定のないジョブ

重大度：情報

無限実行を防ぐため、ジョブにはタイムアウトを設定してください

メッセージ：明示的なタイムアウトを指定せずにジョブが追加されています。ジョブが停止したままになるのを防ぐため、タイムアウトの追加を検討してください。

バックオフ戦略のないリトライ

重大度：警告

リトライには指数バックオフを使用し、一斉集中を避けてください

メッセージ：リトライ回数が設定されていますが、バックオフ戦略がありません。一斉集中を防ぐため、指数バックオフを使用してください。

タイムゾーンが明示されていない繰り返しジョブ

重大度：警告

夏時間の問題を避けるため、繰り返しジョブにはタイムゾーンを指定してください

メッセージ：繰り返しジョブに明示的なタイムゾーンが指定されていません。サーバーのローカル時刻が使用されるため、夏時間によってずれる可能性があります。

高すぎる可能性があるワーカーの並行実行数

重大度：情報

並行実行数が多いと、下流サービスに過剰な負荷がかかる可能性があります

メッセージ：ワーカーの並行実行数が多くなっています。下流サービスがこの負荷（DB 接続、API レート制限）に対応できることを確認してください。

コラボレーション

委任のトリガー

• redis infrastructure|redis cluster|memory tuning -> redis-specialist（キューに Redis インフラストラクチャが必要）
• serverless queue|edge queue|no redis -> upstash-qstash（Redis を管理せずにキューを使用する必要がある）
• complex workflow|saga|compensation|long-running -> temporal-craftsman（単純なジョブを超えるワークフローオーケストレーションが必要）
• event sourcing|CQRS|event streaming -> event-architect（イベント駆動型アーキテクチャが必要）
• deploy|kubernetes|scaling|infrastructure -> devops（キューにインフラストラクチャが必要）
• monitor|metrics|alerting|dashboard -> performance-hunter（キューに監視が必要）

メールキュースタック

スキル：bullmq-specialist、email-systems、redis-specialist

ワークフロー：

1. Email request received (API)
2. Job queued with rate limiting (bullmq-specialist)
3. Worker processes with backoff (bullmq-specialist)
4. Email sent via provider (email-systems)
5. Status tracked in Redis (redis-specialist)

バックグラウンド処理スタック

スキル：bullmq-specialist、backend、devops

ワークフロー：

1. API receives request (backend)
2. Long task queued for background (bullmq-specialist)
3. Worker processes async (bullmq-specialist)
4. Result stored/notified (backend)
5. Workers scaled per load (devops)

AI 処理パイプライン

スキル：bullmq-specialist、ai-workflow-automation、performance-hunter

ワークフロー：

1. AI task submitted (ai-workflow-automation)
2. Job flow created with dependencies (bullmq-specialist)
3. Workers process stages (bullmq-specialist)
4. Performance monitored (performance-hunter)
5. Results aggregated (ai-workflow-automation)

スケジュールタスクスタック

スキル：bullmq-specialist、backend、redis-specialist

ワークフロー：

1. Repeatable jobs defined (bullmq-specialist)
2. Cron patterns with timezone (bullmq-specialist)
3. Jobs execute on schedule (bullmq-specialist)
4. State managed in Redis (redis-specialist)
5. Results handled (backend)

関連スキル

相性のよいスキル：redis-specialist、backend、nextjs-app-router、email-systems、ai-workflow-automation、performance-hunter

使用するタイミング

• ユーザーが bullmq に言及、または暗に示している
• ユーザーが bull queue に言及、または暗に示している
• ユーザーが redis queue に言及、または暗に示している
• ユーザーが background job に言及、または暗に示している
• ユーザーが job queue に言及、または暗に示している
• ユーザーが delayed job に言及、または暗に示している
• ユーザーが repeatable job に言及、または暗に示している
• ユーザーが worker process に言及、または暗に示している
• ユーザーが job scheduling に言及、または暗に示している
• ユーザーが async processing に言及、または暗に示している

制限事項

• タスクが上記の範囲に明確に該当する場合に限り、このスキルを使用してください。
• 出力を、環境固有の検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を止めて確認を求めてください。