agents-inc/skills

BullMQ 模式

快速指南： 使用 BullMQ (v5.x) 在 Redis 之上进行后台作业处理、任务调度和工作流编排。核心类：Queue (添加作业)、Worker (处理作业)、QueueEvents (全局事件监听器)、FlowProducer (父子作业树)。每次构造函数都必须传递 connection 对象 (v5 必需)。对于 Worker 使用的 ioredis 连接，必须设置 maxRetriesPerRequest: null。使用 upsertJobScheduler 实现可重复/定时 (cron) 作业 (替代已弃用的 repeat 选项 API)。QueueScheduler 在 v4 中已被移除——其职责现在由 Worker 自动处理。

---

<critical_requirements>

关键要求：使用此技能前必读

所有代码必须遵循 CLAUDE.md 中的项目约定 (kebab-case、具名导出、导入排序、import type、具名常量)

(你必须向每个 Queue、Worker、QueueEvents 和 FlowProducer 构造函数传递 connection 对象——BullMQ v5 在缺少连接时会抛出错误)

(你必须为 Worker 使用的 ioredis 连接设置 maxRetriesPerRequest: null——BullMQ 需要无限重试，没有此设置会抛出错误)

(你必须在 SIGTERM/SIGINT 信号时调用 await worker.close() 以优雅关闭——否则，正在进行的作业可能变为停滞状态)

(你必须使用 upsertJobScheduler 实现可重复/定时 (cron) 作业——queue.add 上的旧 repeat 选项自 v5.16.0 起已弃用)

</critical_requirements>

---

示例

• 核心模式 -- 队列设置、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 密集型工作

涵盖的关键模式：

• 使用类型化作业数据和返回值的 Queue 和 Worker 设置
• 为 Worker 设置 maxRetriesPerRequest: null 的连接工厂
• 作业选项：延迟、优先级、重试次数、退避、removeOnComplete/Fail
• 使用 worker.close() 根据进程信号优雅关闭
• 用于父子作业树和依赖跟踪的 FlowProducer
• 用于可重复/定时 (cron) 作业的作业调度器 (upsertJobScheduler)
• 速率限制 (全局限速器和手动 Worker.RateLimitError)
• 并发控制 (每个 Worker 本地和全局)
• 用于跨所有 Worker 进行全局事件监控的 QueueEvents
• 用于 CPU 密集型工作的沙箱处理器

何时不使用：

• 简单的进程内定时器或 setTimeout (无需持久化)
• 无需持久化的实时发布/订阅消息传递 (使用你的发布/订阅方案)
• 必须在请求-响应周期内同步处理的数据
• 不需要持久化、重试或调度的队列

---

理念

BullMQ 是一个基于 Redis 的作业队列，用于 Node.js，提供可靠的至少一次交付保证的后台处理。核心原则是：将作业生产与作业消费分离，使应用程序保持响应，同时异步执行工作。

核心原则：

1. 生产者和消费者是解耦的 -- 任何进程都可以将作业添加到队列；任何 Worker 都可以处理它们。这通过添加更多 Worker 实现水平扩展。
2. 作业是持久的 -- 由于作业存储在 Redis 中，因此它们在进程重启后仍然存在。已崩溃的 Worker 的作业会被其他 Worker (或同一 Worker 重启后) 接手。
3. 至少一次交付 -- BullMQ 保证每个作业至少被处理一次。使用幂等的处理器来处理 (罕见的) 停滞后的重复处理情况。
4. 优雅地失败并重试 -- 配置 attempts 和 backoff 策略，以便临时故障自动解决。永久失败的作业会移至失败集合以供检查。
5. 每个 Queue/Worker/QueueEvents 都需要自己的连接 -- BullMQ 在内部管理连接状态。切勿在多个 BullMQ 类之间共享单个 ioredis 实例 (仅作为生产者的 Queues 除外)。

---

核心模式

模式 1：连接工厂

BullMQ v5 要求在每个构造函数上显式提供 Redis 连接。Worker 需要设置 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：队列和 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() 以在退出前完成正在进行的作业。否则，作业将变为停滞状态，并由其他 Worker 重新处理。

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 处理多个 Worker，信号处理器覆盖 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 的间隔会在 0 秒、2 秒、4 秒等时刻触发。

查看 examples/advanced.md 获取移除调度器和列出活动调度器的信息。

---

模式 7：速率限制

限制 Worker 在给定时间窗口内处理的作业数量。限速器对于同一队列上的所有 Worker 是全局生效的。

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 个具有此配置的 Worker 仍然总共每秒最多处理 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}`);
});

为什么好： 从单个监听器监控队列上的所有 Worker，通过 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) 可以共享连接。Worker 和 QueueEvents 由于阻塞 Redis 命令，各自需要自己的连接。

---

<decision_framework>

决策框架

何时使用 BullMQ

是否需要后台作业处理？
|-- 否 -> 不使用 BullMQ
+-- 是 -> 是否需要持久化、重试或调度？
    |-- 否 -> 简单的进程内队列或 setTimeout 可能足够
    +-- 是 -> 是否需要父子作业依赖关系？
        |-- 是 -> BullMQ 配合 FlowProducer
        +-- 否 -> 是否需要速率限制或优先级？
            |-- 是 -> BullMQ 配合限速器/优先级选项
            +-- 否 -> BullMQ 配合基本的 Queue + Worker

选择哪个作业模式？

需要什么样的作业调度？
|-- 一次性延迟作业 -> 使用 delay 选项的 queue.add()

...

[为简洁起见已截断]