bullmq

BullMQ

Node.js 向けの Redis ベースのキューシステムです。中核となるクラスは Queue、Worker、QueueEvents、FlowProducer の 4 つです。

目次

• インストール
• クイックスタート
• 接続
• キュー
• ワーカー
• TypeScript のジェネリクス
• イベント
• ジョブのライフサイクル状態
• 高度なトピック

インストール

yarn add bullmq — Redis 5.0+ が必要です。また、maxmemory-policy=noeviction を設定してください。

<quick_reference>

クイックスタート

import { Queue, Worker, QueueEvents } from "bullmq";

// --- Producer ---
const queue = new Queue("my-queue", {
  connection: { host: "localhost", port: 6379 },
});

await queue.add("job-name", { foo: "bar" });

// --- Consumer ---
const worker = new Worker(
  "my-queue",
  async (job) => {
    // process job
    await job.updateProgress(50);
    return { result: "done" };
  },
  { connection: { host: "localhost", port: 6379 } },
);

worker.on("completed", (job, returnvalue) => {
  console.log(`${job.id} completed with`, returnvalue);
});

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

// IMPORTANT: always attach an error handler
worker.on("error", (err) => {
  console.error(err);
});

// --- Global event listener (all workers) ---
const queueEvents = new QueueEvents("my-queue", {
  connection: { host: "localhost", port: 6379 },
});

queueEvents.on("completed", ({ jobId, returnvalue }) => {
  console.log(`Job ${jobId} completed`);
});

queueEvents.on("failed", ({ jobId, failedReason }) => {
  console.error(`Job ${jobId} failed: ${failedReason}`);
});

ジョブのライフサイクル状態

add() → wait / prioritized / delayed
         ↓
       active → completed
         ↓
       failed → (retry) → wait/delayed

FlowProducer を使用すると、すべての子ジョブが完了するまで、ジョブは waiting-children 状態になることもあります。

</quick_reference>

接続

BullMQ は内部で ioredis を使用します。connection オプションまたは既存の ioredis インスタンスを渡してください。

import { Queue, Worker } from "bullmq";
import { Redis } from "ioredis";

// Option 1: connection config (new connection per instance)
const queue = new Queue("q", {
  connection: { host: "redis.example.com", port: 6379 },
});

// Option 2: reuse ioredis instance (Queue and multiple Queues can share)
const connection = new Redis();
const q1 = new Queue("q1", { connection });
const q2 = new Queue("q2", { connection });

// Option 3: reuse for Workers (BullMQ internally duplicates for blocking)
const workerConn = new Redis({ maxRetriesPerRequest: null });
const w1 = new Worker("q1", async (job) => {}, { connection: workerConn });

重要なルール：

• ワーカーでは、ioredis インスタンスに maxRetriesPerRequest: null を設定する必要があります。BullMQ はこの設定を強制しており、設定されていない場合は警告またはエラーを出します。
• ioredis の keyPrefix オプションは使用しないでください。代わりに BullMQ の prefix オプションを使用してください。
• QueueEvents はブロッキング Redis コマンドを使用するため、接続を共有できません。
• Redis には maxmemory-policy=noeviction を設定する必要があります。

キュー

const queue = new Queue("paint", { connection });

// Add a job
await queue.add("job-name", { color: "red" });

// Add with options
await queue.add(
  "job-name",
  { color: "blue" },
  {
    delay: 5000, // wait 5s before processing
    priority: 1, // lower = higher priority (0 is highest, max 2^21)
    attempts: 3, // retry up to 3 times
    backoff: { type: "exponential", delay: 1000 },
    removeOnComplete: true, // or { count: 100 } to keep last 100
    removeOnFail: 1000, // keep last 1000 failed jobs
  },
);

// Add bulk
await queue.addBulk([
  { name: "job1", data: { x: 1 } },
  { name: "job2", data: { x: 2 }, opts: { priority: 1 } },
]);

// Queue operations
await queue.pause();
await queue.resume();
await queue.obliterate({ force: true }); // remove all data
await queue.close();

ワーカー

const worker = new Worker<MyData, MyReturn>(
  "paint",
  async (job) => {
    await job.updateProgress(42);
    return { cost: 100 };
  },
  {
    connection,
    concurrency: 5, // process 5 jobs concurrently
    autorun: false, // don't start immediately
  },
);

worker.run(); // start when ready

// Update concurrency at runtime
worker.concurrency = 10;

プロセッサーが受け取る引数は 3 つです： (job, token?, signal?) — signal はキャンセルをサポートする AbortSignal です。

TypeScript のジェネリクス

interface JobData {
  color: string;
}
interface JobReturn {
  cost: number;
}

const queue = new Queue<JobData, JobReturn>("paint");
const worker = new Worker<JobData, JobReturn>("paint", async (job) => {
  // job.data is typed as JobData
  return { cost: 100 }; // must match JobReturn
});

イベント

ワーカーイベント（そのワーカーインスタンス内で発生）：

イベント	コールバックのシグネチャ
completed	(job, returnvalue)
failed	(job | undefined, error, prev)
progress	(job, progress: number | object)
drained	() — キューが空です
error	(error) — このハンドラーを必ず登録してください

QueueEvents（すべてのワーカーを対象とするグローバルイベント。Redis Streams を使用）：

イベント	コールバックのシグネチャ
completed	({ jobId, returnvalue })
failed	({ jobId, failedReason })
progress	({ jobId, data })
waiting	({ jobId })
active	({ jobId, prev })
delayed	({ jobId, delay })
deduplicated	({ jobId, deduplicationId, deduplicatedJobId })

イベントストリームは約 10,000 件のイベントに自動でトリミングされます。streams.events.maxLen で設定できます。

高度なトピック

• ジョブの種類とオプション（遅延、優先順位付け、重複排除、繰り返し）：references/job-types-and-options.md を参照してください
• フローとスケジューラー（FlowProducer、親子関係、ジョブスケジューラー、cron）：references/flows-and-schedulers.md を参照してください
• パターン（ステップジョブ、冪等性、スロットリング、手動レート制限）：references/patterns.md を参照してください
• 本番環境（シャットダウン、Redis の設定、リトライ、バックオフ、モニタリング）：references/production.md を参照してください