agents-inc/skills

BullMQ 模式

快速指南： 使用 BullMQ (v5.x) 在 Redis 之上進行背景任務處理、任務排程與工作流程編排。核心類別：Queue（加入任務）、Worker（處理任務）、QueueEvents（全域事件監聽器）、FlowProducer（父子任務樹）。務必將 connection 物件傳遞給每個建構函式（v5 的必要條件）。在 Worker 使用的 ioredis 連線上設定 maxRetriesPerRequest: null。使用 upsertJobScheduler 來處理可重複/cron 任務（取代已棄用的 repeatable API）。QueueScheduler 已在 v4 中移除——其職責現已自動由 Workers 處理。

---

<critical_requirements>

重要事項：使用此技能前

所有程式碼必須遵循 CLAUDE.md 中的專案慣例（kebab-case、具名匯出、匯入順序、import type、具名常數）

（你必須將 connection 物件傳遞給每個 Queue、Worker、QueueEvents 和 FlowProducer 建構函式——BullMQ v5 在缺少連線時會擲出錯誤）

（你必須在 Workers 使用的 ioredis 連線上設定 maxRetriesPerRequest: null——BullMQ 需要無限重試，沒有此設定會擲出錯誤）

（你必須在 SIGTERM/SIGINT 時呼叫 await worker.close() 以優雅關機——否則進行中的任務會變成停滯狀態）

（你必須使用 upsertJobScheduler 處理可重複/cron 任務——自 v5.16.0 起，queue.add 上的舊 repeat 選項已棄用）

</critical_requirements>

---

範例

• 核心模式 —— Queue 設定、Worker 處理、任務選項、連線工廠、優雅關機、類型化任務
• 進階模式 —— FlowProducer、速率限制、任務排程、QueueEvents、並行、沙盒處理器

其他資源：

• reference.md —— 決策框架、任務選項參考、反模式、生產環境檢查清單

---

自動偵測： BullMQ, bullmq, Queue, Worker, QueueEvents, FlowProducer, job queue, background job, worker process, job scheduler, upsertJobScheduler, rate limiter, job priority, job delay, sandboxed processor, repeatable job, cron job, flow producer, parent child jobs

使用時機：

• 背景處理（發送電子郵件、圖片處理、PDF 產生）
• 排程/cron 任務（夜間報表、定期清理）
• 具有父子任務依賴關係的工作流程編排（FlowProducer）
• 受速率限制的 API 消費（節流外送請求）
• 基於優先權的任務處理（緊急任務優先於批次操作）
• 將 CPU 密集型工作分散到多個 Worker 或機器

涵蓋的關鍵模式：

• 使用類型化任務資料和回傳值的 Queue 與 Worker 設定
• 為 Workers 設定的連線工廠（包含 maxRetriesPerRequest: null）
• 任務選項：延遲、優先權、重試次數、退避、removeOnComplete/Fail
• 透過 worker.close() 在程序訊號觸發時優雅關機
• 使用 FlowProducer 建立具有依賴追蹤的父子任務樹
• 任務排程器用於可重複/cron 任務 (upsertJobScheduler)
• 速率限制（全域限制器和手動 Worker.RateLimitError）
• 並行控制（每個 Worker 的本機和全域）
• 使用 QueueEvents 進行所有 Worker 的全域事件監控
• 用於 CPU 密集型工作的沙盒處理器

何時不該使用：

• 簡單的進程內計時器或 setTimeout（不需要持久化）
• 不需要持久化的即時發布/訂閱訊息傳遞（使用你的發布/訂閱解決方案）
• 必須在請求-回應週期內同步處理的資料
• 不需要持久化、重試或排程的佇列

---

理念

BullMQ 是一個基於 Redis 的任務佇列，適用於 Node.js，提供可靠的背景處理，具有至少一次的傳遞保證。核心原則：將任務產生與任務消費分離，使你的應用程式保持回應能力，同時非同步執行工作。

核心原則：

1. 生產者和消費者解耦 —— 任何程序都可以將任務加入佇列；任何 Worker 都可以處理它們。這可以透過增加更多 Workers 來實現水平擴展。
2. 任務是持久的 —— 任務在 Redis 中存活，即使程序重新啟動也不會遺失。崩潰的 Worker 的任務會由其他 Workers（或重新啟動後的同一個 Worker）接手。
3. 至少一次傳遞 —— BullMQ 保證每個任務至少被處理一次。使用冪等處理器來處理（罕見的）因停滯而重複處理的情況。
4. 優雅地失敗並重試 —— 設定 attempts 和 backoff 策略，以便暫時性故障能自動解決。永久失敗的任務會移至失敗集合以供檢查。
5. 每個 Queue/Worker/QueueEvents 都需要自己的連線 —— BullMQ 在內部管理連線狀態。切勿在單個 BullMQ 類別之間共用單個 ioredis 實例（僅作為生產者的 Queue 除外）。

---

核心模式

模式 1：連線工廠

BullMQ v5 需要在每個建構函式上提供明確的 Redis 連線。Workers 需要 maxRetriesPerRequest: null 以便 ioredis 無限重試，而不是放棄。

import Redis from "ioredis";

function createBullMQConnection(): Redis {
  const url = process.env.REDIS_URL;
  if (!url) throw new Error("REDIS_URL is required");
  return new Redis(url, { maxRetriesPerRequest: null });
}
export { createBullMQConnection };

為何良好： maxRetriesPerRequest: null 滿足 BullMQ 的要求，工廠函式確保一致的設定，環境變數將憑證保留在程式碼之外

// 錯誤 —— 缺少 maxRetriesPerRequest
const redis = new Redis("redis://localhost:6379");
const worker = new Worker("emails", processor, { connection: redis });
// BullMQ 擲出錯誤："maxRetriesPerRequest must be null"

為何錯誤： BullMQ 在 Worker 連線上需要無限重試，若沒有 null 設定，啟動時會擲出錯誤

請參閱 examples/core.md 以了解包含生產者與消費者分離的完整連線工廠。

---

模式 2：Queue 與 Worker 設定

Queue 加入任務；Worker 處理任務。兩者都接受 TypeScript 泛型以實現類型安全的任務資料和回傳值。

import { Queue, Worker, type Job } from "bullmq";

interface EmailJobData {
  to: string;
  subject: string;
  body: string;
}

const QUEUE_NAME = "emails";

const emailQueue = new Queue<EmailJobData>(QUEUE_NAME, {
  connection: createBullMQConnection(),
});

const emailWorker = new Worker<EmailJobData>(
  QUEUE_NAME,
  async (job: Job<EmailJobData>) => {
    await sendEmail(job.data.to, job.data.subject, job.data.body);
  },
  { connection: createBullMQConnection() },
);

為何良好： 泛型強制 job.data 的類型安全，Queue 和 Worker 使用各自的連線，使用具名常數表示佇列名稱

請參閱 examples/core.md 以了解包含事件、錯誤處理和類型化回傳值的完整設定。

---

模式 3：任務選項

使用選項控制任務行為：延遲、優先權、重試、退避和自動移除。

const MAX_ATTEMPTS = 5;
const BACKOFF_DELAY_MS = 1000;
const DELAY_MS = 60_000;
const KEEP_COMPLETED_COUNT = 100;

await emailQueue.add(
  "send-welcome",
  { to: "[email protected]", subject: "Welcome", body: "..." },
  {
    delay: DELAY_MS,
    priority: 1, // 1 = 最高，數字越大 = 優先權越低
    attempts: MAX_ATTEMPTS,
    backoff: { type: "exponential", delay: BACKOFF_DELAY_MS },
    removeOnComplete: { count: KEEP_COMPLETED_COUNT },
    removeOnFail: false,
  },
);

為何良好： 所有數值使用具名常數，針對暫時性失敗使用指數退避，使用 count 的 removeOnComplete 防止無限增加 Redis 記憶體，removeOnFail: false 保留失敗任務以便除錯

請參閱 examples/core.md 以了解 defaultJobOptions、addBulk 和 LIFO 模式。

---

模式 4：優雅關機

在程序訊號觸發時呼叫 worker.close()，以便在退出前完成進行中的任務。如果不這樣做，任務會變成停滯狀態，並由其他 Workers 重新處理。

async function shutdown(workers: Worker[]): Promise<void> {
  await Promise.all(workers.map((w) => w.close()));
  process.exit(0);
}

process.on("SIGTERM", () => shutdown([emailWorker]));
process.on("SIGINT", () => shutdown([emailWorker]));

為何良好： close() 停止接受新任務並等待進行中的任務完成，Promise.all 處理多個 Workers，訊號處理器涵蓋 SIGTERM 和 SIGINT

陷阱： worker.close() 沒有內建超時。如果處理器掛起，關機也會掛起。如果需要，請自行封裝超時。

請參閱 examples/core.md 以了解包含超時模式的關機。

---

模式 5：FlowProducer（父子任務）

FlowProducer 建立任務樹，其中父任務等待所有子任務完成。整個樹是原子性地加入的。

import { FlowProducer } from "bullmq";

const flowProducer = new FlowProducer({ connection: createBullMQConnection() });

await flowProducer.add({
  name: "publish-video",
  queueName: "videos",
  children: [
    { name: "extract-audio", queueName: "media", data: { videoId: "abc" } },
    {
      name: "generate-thumbnail",
      queueName: "media",
      data: { videoId: "abc" },
    },
    {
      name: "transcode",
      queueName: "media",
      data: { videoId: "abc", format: "mp4" },
    },
  ],
});

為何良好： 父任務等待所有子任務完成，原子性加入（全有或全無），子任務可以位於不同的佇列中，父處理器可以透過 job.getChildrenValues() 存取子任務的結果

請參閱 examples/advanced.md 以了解如何存取子任務值與巢狀流程。

---

模式 6：任務排程器（可重複/Cron 任務）

使用 upsertJobScheduler（v5.16.0+）處理可重複的任務。取代已棄用的 repeat 選項。

const REPORT_INTERVAL_MS = 3_600_000; // 1 小時

// 固定間隔
await emailQueue.upsertJobScheduler(
  "hourly-digest",
  { every: REPORT_INTERVAL_MS },
  {
    name: "send-digest",
    data: { type: "hourly" },
  },
);

// Cron 表達式 —— 每天凌晨 3:15
await emailQueue.upsertJobScheduler(
  "nightly-cleanup",
  { pattern: "0 15 3 * * *" },
  {
    name: "cleanup",
    data: { type: "nightly" },
    opts: { removeOnComplete: true },
  },
);

為何良好： upsertJobScheduler 是冪等的（在每次啟動時呼叫是安全的），範本定義了每個產生的任務繼承的任務名稱/資料/選項，透過 pattern 使用 cron 語法

陷阱： every 間隔是對齊時鐘的，而不是對齊你呼叫 upsertJobScheduler 的時間。一個 2000ms 的間隔會在 0s、2s、4s 等時間點觸發。

請參閱 examples/advanced.md 以了解如何移除排程器和列出啟用中的排程器。

---

模式 7：速率限制

限制 Worker 在每個時間視窗內處理的任務數量。限制器是全域的，適用於同一個佇列上的所有 Workers。

const RATE_LIMIT_MAX = 10;
const RATE_LIMIT_DURATION_MS = 1000;

const worker = new Worker("api-calls", processor, {
  connection: createBullMQConnection(),
  limiter: { max: RATE_LIMIT_MAX, duration: RATE_LIMIT_DURATION_MS },
});

為何良好： 全域速率限制（具有此設定的 10 個 Workers 總計每秒仍最多處理 10 個任務），使用具名常數表示限制

對於基於外部 API 回應的動態速率限制，請使用 worker.rateLimit(duration) + throw Worker.RateLimitError()。請參閱 examples/advanced.md。

---

模式 8：使用 QueueEvents 進行全域監控

QueueEvents 使用 Redis Streams（而非 Pub/Sub），因此事件即使在重新連線時也是可靠的。需要自己的專用連線。

import { QueueEvents } from "bullmq";

const queueEvents = new QueueEvents(QUEUE_NAME, {
  connection: createBullMQConnection(),
});

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

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

為何良好： 從單一監聽器監控佇列上的所有 Workers，透過 Redis Streams 實現可靠傳遞，按需求使用各自的連線

請參閱 examples/advanced.md 以了解進度追蹤和等待特定任務完成。

---

效能最佳化

• 並行 —— 在 Worker 選項中設定 concurrency，以在單個 Worker 實例內並行處理多個任務：{ concurrency: 10 }。僅對 I/O 密集型工作有效。CPU 密集型工作會阻塞事件循環並導致任務停滯——改用沙盒處理器。
• 沙盒處理器 —— 將檔案路徑而不是函式傳遞給 Worker 建構函式，以便在單獨的進程（或使用 useWorkerThreads: true 的工作執行緒）中執行處理器。防止 CPU 密集型工作阻塞鎖定續約。
• addBulk —— 使用 queue.addBulk([...]) 在單次 Redis 往返中新增多個任務，而不是在迴圈中呼叫 queue.add()。
• removeOnComplete/Fail —— 設定自動移除以防止 Redis 記憶體無限增長。使用 { count: N } 保留最後 N 個任務以供除錯。
• 分離連線 —— 生產者（Queue）可以共用連線。Workers 和 QueueEvents 每個都需要自己的連線，因為它們會使用阻塞的 Redis 命令。

---

<decision_framework>

決策框架

何時使用 BullMQ

你需要背景任務處理嗎？
|-- 否 -> 不要使用 BullMQ
+-- 是 -> 你需要持久化、重試或排程嗎？
    |-- 否 -> 簡單的進程內佇列或 setTimeout 可能就足夠了
    +-- 是 -> 你需要父子任務依賴關係嗎？
        |-- 是 -> BullMQ 搭配 FlowProducer
        +-- 否 -> 你需要速率限制或優先權嗎？
            |-- 是 -> BullMQ 搭配 limiter/priority 選項
            +-- 否 -> BullMQ 使用基本的 Queue + Worker

使用哪種任務模式？

你需要哪種任務排程？
|-- 一次性延遲任務 -> 使用具有 delay 選項的 queue.add()
|-- 可重複的間隔任務 -> 使用具有 `every` 選項的 upsertJobScheduler
|-- 可重複的 cron 任務 -> 使用具有 `pattern` 選項的 upsertJobScheduler
+-- 具有依賴關係的工作流程 -> 使用 FlowProducer

生產就緒清單

• 所有 Workers 的 Redis 連線都設定了 maxRetriesPerRequest: null
• 優雅關機已實作（Worker.close() 搭配 SIGTERM/SIGINT 處理器）
• 任務處理器是冪等的（可能收到重複的任務）
• 為生產者（Queue）和消費者（Worker/QueueEvents）使用分離的連線
• removeOnComplete 設定為防止佇列無限增長（使用 count 或 age）
• 使用 try-catch 或全域事件監聽器處理失敗的任務
• 使用 upsertJobScheduler 而不是舊的 repeat 選項
• Redis 在生產環境中啟用了持久化（至少為 AOF）
• 已為生產 Worker 設定並行（concurrency）
• 已為 CPU 密集型任務評估沙盒處理器（檔案路徑）
• 使用 queue.addBulk() 進行批次任務新增
• 任務資料已序列化（僅限 JSON 可序列化的資料）
• 佇列名稱是常數（不是內聯字串）
• attempts 和 backoff 已根據你的失敗容忍度進行設定

</decision_framework>