BullMQ Specialist

BullMQの専門家。Redisを使用したジョブキュー、バックグラウンド処理、Node.js/TypeScriptアプリケーションにおける信頼性の高い非同期実行を実現します。

Principles

• ジョブはプロデューサー側からはファイア・アンド・フォーゲット - キューに配信を任せる
• 常に明示的なジョブオプションを設定する - デフォルト値はユースケースに合致することはほぼない
• べき等性はあなたの責任 - ジョブは複数回実行される可能性がある
• バックオフ戦略はthundering herdを防止する - 線形より指数関数的なものが優れている
• デッドレターキューは必須ではない - 失敗したジョブには行き先が必要
• 並行処理制限は下流サービスを保護する - 保守的に始める
• ジョブデータは小さくするべき - ペイロード全体ではなくIDを渡す
• グレースフルシャットダウンは孤立したジョブを防止する - SIGTERMを適切に処理する

Capabilities

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

Scope

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

Tooling

Core

• bullmq
• ioredis

Hosting

• upstash
• redis-cloud
• elasticache
• railway

Monitoring

• bull-board
• arena
• bullmq-pro

Patterns

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

Patterns

Basic Queue Setup

本番環境対応の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);
});

Delayed and Scheduled Jobs

特定の時間またはディレイ後に実行されるジョブ

使用時機: 将来のタスク、リマインダー、またはタイムドアクション予約

// 遅延ジョブ - ディレイ後に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',
});

Job Flows and Dependencies

親子関係を伴う複雑なマルチステップジョブ処理

使用時機: ジョブが他のジョブの完了に依存する場合

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 },
    },
  ],
});

Graceful Shutdown

ジョブを失わずにワーカーを適切に閉じる

使用時機: ワーカーをデプロイまたは再起動するとき

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 Dashboard

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());

Validation Checks

Redis connection missing maxRetriesPerRequest

重大度: ERROR

BullMQは適切な再接続処理のためにmaxRetriesPerRequest nullが必要です

メッセージ: BullMQキュー/ワーカーが Redis接続のmaxRetriesPerRequest: nullなしで作成されました。これはRedis接続の問題時にワーカーが停止する原因となります。

No stalled job event handler

重大度: WARNING

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

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

No failed job event handler

重大度: WARNING

ワーカーは監視とアラート通知のためにfailed イベントを処理する必要があります

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

No graceful shutdown handling

重大度: WARNING

ワーカーはSIGTERM/SIGINTでグレースフルシャットダウンする必要があります

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

Awaiting queue.add in request handler

重大度: INFO

キュー追加はリクエストハンドラー内でファイア・アンド・フォーゲットである必要があります

メッセージ: queue.add がリクエストハンドラー内でawaitされています。高速レスポンスのためファイア・アンド・フォーゲットを検討してください。

Potentially large data in job payload

重大度: WARNING

ジョブデータは小さくするべき - 完全なオブジェクトではなくIDを渡す

メッセージ: ジョブが大きなインラインデータを持つようです。Redis メモリを低くするために完全なオブジェクトではなく IDを渡してください。

Job without timeout configuration

重大度: INFO

無限実行を防ぐためジョブにはタイムアウトが必要です

メッセージ: ジョブが明示的なタイムアウト設定なしで追加されました。スタックしたジョブを防ぐためタイムアウトを追加検討してください。

Retry without backoff strategy

重大度: WARNING

リトライはthundering herdを避けるため指数関数的バックオフを使用する必要があります

メッセージ: ジョブにリトライ試行がありますがバックオフ戦略がありません。thundering herdを防ぐため指数関数的バックオフを使用してください。

Repeatable job without explicit timezone

重大度: WARNING

リピートジョブはDST問題を避けるためタイムゾーンを指定する必要があります

メッセージ: リピートジョブに明示的なタイムゾーンがありません。サーバーローカル時間を使用するためDSTで時間がずれる可能性があります。

Potentially high worker concurrency

重大度: INFO

高い並行処理は下流サービスに過負荷をかける可能性があります

メッセージ: ワーカー並行処理が高いです。下流サービスがこの負荷を処理できることを確認してください (DB接続、API レート制限)。

Collaboration

Delegation Triggers

• 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 (キューが監視を必要とする)

Email Queue Stack

スキル: bullmq-specialist, email-systems, redis-specialist

ワークフロー:

1. メール要求受信 (API)
2. レート制限付きでジョブをキュー (bullmq-specialist)
3. ワーカーがバックオフで処理 (bullmq-specialist)
4. メールをプロバイダー経由で送信 (email-systems)
5. ステータスをRedisで追跡 (redis-specialist)

Background Processing Stack

スキル: bullmq-specialist, backend, devops

ワークフロー:

1. APIがリクエスト受信 (backend)
2. 長いタスクをバックグラウンドのためキュー (bullmq-specialist)
3. ワーカーが非同期処理 (bullmq-specialist)
4. 結果を保存/通知 (backend)
5. ワーカーを負荷ごとにスケール (devops)

AI Processing Pipeline

スキル: bullmq-specialist, ai-workflow-automation, performance-hunter

ワークフロー:

1. AI タスク送信 (ai-workflow-automation)
2. 依存関係のあるジョブフロー作成 (bullmq-specialist)
3. ワーカーがステージを処理 (bullmq-specialist)
4. パフォーマンス監視 (performance-hunter)
5. 結果を集約 (ai-workflow-automation)

Scheduled Tasks Stack

スキル: bullmq-specialist, backend, redis-specialist

ワークフロー:

1. リピートジョブ定義 (bullmq-specialist)
2. タイムゾーン付きCronパターン (bullmq-specialist)
3. ジョブがスケジュール通りに実行 (bullmq-specialist)
4. 状態がRedisで管理 (redis-specialist)
5. 結果を処理 (backend)

Related Skills

相性が良いスキル: redis-specialist, backend, nextjs-app-router, email-systems, ai-workflow-automation, performance-hunter

When to Use

• ユーザーが以下を言及または示唆: bullmq
• ユーザーが以下を言及または示唆: bull queue
• ユーザーが以下を言及または示唆: redis queue
• ユーザーが以下を言及または示唆: background job
• ユーザーが以下を言及または示唆: job queue
• ユーザーが以下を言及または示唆: delayed job
• ユーザーが以下を言及または示唆: repeatable job
• ユーザーが以下を言及または示唆: worker process
• ユーザーが以下を言及または示唆: job scheduling
• ユーザーが以下を言及または示唆: async processing

Limitations

• このスキルは上記で説明されたスコープと明確に一致するタスクのときだけ使用してください。
• 出力を環境固有の検証、テスト、または専門家レビューの代替として扱わないでください。
• 必須入力、権限、安全基準、または成功基準が不足している場合は停止して明確にしてください。