Back to Details

api-queue-bullmq

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

BullMQ Patterns

BullMQ 模式

Quick Guide: Use BullMQ (v5.x) for background job processing, task scheduling, and workflow orchestration on top of Redis. Core classes: 
Queue
(adds jobs), 
Worker
(processes jobs), 
QueueEvents
(global event listener), 
FlowProducer
(parent-child job trees). Always pass a 
connection
object to every constructor (required in v5). Set 
maxRetriesPerRequest: null
on ioredis connections for Workers. Use 
upsertJobScheduler
for repeatable/cron jobs (replaces deprecated repeatable API). QueueScheduler was removed in v4 -- its responsibilities are now handled by Workers automatically.
<critical_requirements>
快速指南： 基于Redis，使用BullMQ（v5.x版本）实现后台作业处理、任务调度与工作流编排。核心类：
Queue
（添加作业）、
Worker
（处理作业）、
QueueEvents
（全局事件监听器）、
FlowProducer
（父子作业树）。必须为每个构造函数传入
connection
对象（v5版本必填）。为Worker的ioredis连接设置
maxRetriesPerRequest: null
。使用
upsertJobScheduler
处理重复/定时作业（替代已废弃的repeatable API）。QueueScheduler在v4版本中已移除——其职责现在由Worker自动处理。
<critical_requirements>

CRITICAL: Before Using This Skill

重要提示：使用本技能前须知

All code must follow project conventions in CLAUDE.md (kebab-case, named exports, import ordering, 
import type
, named constants)
(You MUST pass a 
connection
object to every Queue, Worker, QueueEvents, and FlowProducer constructor -- BullMQ v5 throws if connection is missing)
(You MUST set 
maxRetriesPerRequest: null
on ioredis connections used by Workers -- BullMQ requires infinite retries and throws without this setting)
(You MUST call 
await worker.close()
on SIGTERM/SIGINT for graceful shutdown -- without it, in-progress jobs become stalled)
(You MUST use 
upsertJobScheduler
for repeatable/cron jobs -- the old 
repeat
option on 
queue.add
is deprecated since v5.16.0)
</critical_requirements>
所有代码必须遵循CLAUDE.md中的项目规范（短横线命名、命名导出、导入顺序、
import type
、命名常量）
（必须为每个Queue、Worker、QueueEvents和FlowProducer构造函数传入
connection
对象——如果缺少连接，BullMQ v5会抛出错误）
（必须为Worker使用的ioredis连接设置
maxRetriesPerRequest: null
——BullMQ要求无限重试，缺少该设置会抛出错误）
（必须在SIGTERM/SIGINT信号触发时调用
await worker.close()
以优雅关闭——否则，正在处理的作业会陷入停滞状态）
（必须使用
upsertJobScheduler
处理重复/定时作业——v5.16.0版本起，
queue.add
中的旧
repeat
选项已废弃）
</critical_requirements>

Examples

示例

  • Core Patterns -- Queue setup, Worker processing, job options, connection factory, graceful shutdown, typed jobs
  • Advanced Patterns -- FlowProducer, rate limiting, job scheduling, QueueEvents, concurrency, sandboxed processors
Additional resources:
  • reference.md -- Decision frameworks, job option reference, anti-patterns, production checklist
Auto-detection: 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
When to use:
  • Background processing (email sending, image processing, PDF generation)
  • Scheduled/cron jobs (nightly reports, periodic cleanup)
  • Workflow orchestration with parent-child job dependencies (FlowProducer)
  • Rate-limited API consumption (throttling outbound requests)
  • Priority-based job processing (urgent jobs before bulk operations)
  • Distributing CPU-intensive work across multiple workers or machines
Key patterns covered:
  • Queue and Worker setup with typed job data and return values
  • Connection factory with 
    maxRetriesPerRequest: null
    for Workers
  • Job options: delay, priority, attempts, backoff, removeOnComplete/Fail
  • Graceful shutdown with 
    worker.close()
    on process signals
  • FlowProducer for parent-child job trees with dependency tracking
  • Job Schedulers for repeatable/cron jobs (
    upsertJobScheduler
    )
  • Rate limiting (global limiter and manual 
    Worker.RateLimitError
    )
  • Concurrency control (local per-worker and global)
  • QueueEvents for global event monitoring across all workers
  • Sandboxed processors for CPU-intensive work
When NOT to use:
  • Simple in-process timers or 
    setTimeout
    (no persistence needed)
  • Real-time pub/sub messaging without persistence (use your pub/sub solution)
  • Data that must be processed synchronously within a request-response cycle
  • Queues that don't need persistence, retries, or scheduling
<philosophy>
  • 核心模式 —— 队列设置、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调用（限制出站请求）
  • 基于优先级的作业处理（紧急作业优先于批量操作）
  • 跨多个Worker或机器分配CPU密集型任务
涵盖的核心模式：
  • 带类型化作业数据和返回值的Queue与Worker设置
  • 为Worker配置
    maxRetriesPerRequest: null
    的连接工厂
  • 作业选项：延迟、优先级、重试次数、退避策略、完成/失败后自动移除
  • 通过进程信号调用
    worker.close()
    实现优雅关闭
  • 带依赖跟踪的FlowProducer父子作业树
  • 用于重复/定时作业的作业调度器（
    upsertJobScheduler
    ）
  • 速率限制（全局限制器与手动
    Worker.RateLimitError
    ）
  • 并发控制（单Worker本地限制与全局限制）
  • 跨所有Worker的全局事件监控QueueEvents
  • 用于CPU密集型任务的沙箱处理器
不适用场景：
  • 简单的进程内定时器或
    setTimeout
    （无需持久化）
  • 无需持久化的实时发布/订阅消息（使用专用的发布/订阅方案）
  • 必须在请求-响应周期内同步处理的数据
  • 无需持久化、重试或调度的队列
<philosophy>

Philosophy

设计理念

BullMQ is a Redis-backed job queue for Node.js that provides reliable background processing with at-least-once delivery guarantees. The core principle: separate job production from job consumption so your application stays responsive while work happens asynchronously.
Core principles:
  1. Producers and consumers are decoupled -- Any process can add jobs to a queue; any Worker can process them. This enables horizontal scaling by adding more Workers.
  2. Jobs are persistent -- Jobs survive process restarts because they live in Redis. A crashed Worker's jobs are picked up by other Workers (or the same Worker after restart).
  3. At-least-once delivery -- BullMQ guarantees every job is processed at least once. Use idempotent processors to handle the (rare) case of duplicate processing after a stall.
  4. Fail gracefully with retries -- Configure 
    attempts
    and 
    backoff
    strategies so transient failures resolve automatically. Permanently failed jobs move to the failed set for inspection.
  5. Each Queue/Worker/QueueEvents needs its own connection -- BullMQ manages connection state internally. Never share a single ioredis instance across multiple BullMQ classes (except Queues acting only as producers).
</philosophy>
<patterns>
BullMQ是一款基于Redis的Node.js作业队列，提供可靠的后台处理，保证至少一次交付。核心原则：将作业生产与作业消费分离，使应用在异步处理任务时保持响应性。
核心原则：
  1. 生产者与消费者解耦 —— 任意进程都可向队列添加作业；任意Worker都可处理作业。这支持通过添加更多Worker实现水平扩展。
  2. 作业持久化 —— 作业存储在Redis中，可在进程重启后保留。崩溃的Worker的作业会被其他Worker（或重启后的同一Worker）接手处理。
  3. 至少一次交付 —— BullMQ保证每个作业至少被处理一次。使用幂等处理器处理停滞作业可能导致的（罕见）重复处理情况。
  4. 通过重试优雅失败 —— 配置
    attempts
    和退避策略，使临时故障自动恢复。永久失败的作业会移至失败集合供检查。
  5. 每个Queue/Worker/QueueEvents需要独立连接 —— BullMQ内部管理连接状态。切勿在多个BullMQ类之间共享单个ioredis实例（仅作为生产者的Queue除外）。
</philosophy>
<patterns>

Core Patterns

核心模式

Pattern 1: Connection Factory

模式1：连接工厂

BullMQ v5 requires an explicit Redis connection on every constructor. Workers need 
maxRetriesPerRequest: null
so ioredis retries indefinitely instead of giving up.
typescript
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 };
Why good: 
maxRetriesPerRequest: null
satisfies BullMQ's requirement, factory ensures consistent config, environment variable keeps credentials out of code
typescript
// Bad -- missing maxRetriesPerRequest
const redis = new Redis("redis://localhost:6379");
const worker = new Worker("emails", processor, { connection: redis });
// BullMQ throws: "maxRetriesPerRequest must be null"
Why bad: BullMQ requires infinite retries on Worker connections and will throw at startup without 
null
See examples/core.md for the full connection factory with producer vs consumer separation.
BullMQ v5要求每个构造函数都传入显式的Redis连接。Worker需要设置
maxRetriesPerRequest: null
，使ioredis无限重试而非放弃。
typescript
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的要求，工厂确保配置一致，环境变量避免凭证硬编码
typescript
// 错误示例 —— 缺少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获取完整的连接工厂，包含生产者与消费者分离逻辑。

Pattern 2: Queue and Worker Setup

模式2：Queue与Worker设置

Queue adds jobs; Worker processes them. Both accept TypeScript generics for type-safe job data and return values.
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() },
);
Why good: Generics enforce type safety on 
job.data
, separate connections for Queue and Worker, named constant for queue name
See examples/core.md for full setup with events, error handling, and typed return values.
Queue负责添加作业；Worker负责处理作业。两者都支持TypeScript泛型，实现类型安全的作业数据与返回值。
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获取完整设置，包含事件、错误处理和类型化返回值。

Pattern 3: Job Options

模式3：作业选项

Control job behavior with options: delay, priority, retries, backoff, and auto-removal.
typescript
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: "user@example.com", subject: "Welcome", body: "..." },
  {
    delay: DELAY_MS,
    priority: 1, // 1 = highest, higher numbers = lower priority
    attempts: MAX_ATTEMPTS,
    backoff: { type: "exponential", delay: BACKOFF_DELAY_MS },
    removeOnComplete: { count: KEEP_COMPLETED_COUNT },
    removeOnFail: false,
  },
);
Why good: Named constants for all numeric values, exponential backoff for transient failures, 
removeOnComplete
with count prevents unbounded Redis memory, 
removeOnFail: false
preserves failed jobs for debugging
See examples/core.md for 
defaultJobOptions
, 
addBulk
, and LIFO patterns.
通过选项控制作业行为：延迟、优先级、重试、退避和自动移除。
typescript
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: "user@example.com", 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,
  },
);
优势： 所有数值使用命名常量，指数退避处理临时故障，
removeOnComplete
设置数量防止Redis内存无限增长，
removeOnFail: false
保留失败作业用于调试
查看examples/core.md获取
defaultJobOptions
、
addBulk
和LIFO模式。

Pattern 4: Graceful Shutdown

模式4：优雅关闭

Call 
worker.close()
on process signals to finish in-progress jobs before exiting. Without this, jobs become stalled and are re-processed by other Workers.
typescript
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]));
Why good: 
close()
stops accepting new jobs and waits for in-progress jobs to finish, 
Promise.all
handles multiple workers, signal handlers cover both SIGTERM and SIGINT
Gotcha: 
worker.close()
has no built-in timeout. If a processor hangs, the shutdown hangs. Wrap with your own timeout if needed.
See examples/core.md for shutdown with timeout pattern.
在收到进程信号时调用
worker.close()
，在退出前完成正在处理的作业。否则，作业会陷入停滞并被其他Worker重新处理。
typescript
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获取带超时的关闭模式。

Pattern 5: FlowProducer (Parent-Child Jobs)

模式5：FlowProducer（父子作业）

FlowProducer creates job trees where parent jobs wait for all children to complete. The entire tree is added atomically.
typescript
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" },
    },
  ],
});
Why good: Parent waits until all children complete, atomic addition (all or nothing), children can be in different queues, parent processor can access children results via 
job.getChildrenValues()
See examples/advanced.md for accessing children values and nested flows.
FlowProducer创建作业树，父作业等待所有子作业完成。整个作业树会被原子性添加。
typescript
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获取访问子作业值和嵌套流的方法。

Pattern 6: Job Schedulers (Repeatable/Cron Jobs)

模式6：作业调度器（重复/定时作业）

Use 
upsertJobScheduler
(v5.16.0+) for repeatable jobs. Replaces the deprecated 
repeat
option.
typescript
const REPORT_INTERVAL_MS = 3_600_000; // 1 hour

// Fixed interval
await emailQueue.upsertJobScheduler(
  "hourly-digest",
  { every: REPORT_INTERVAL_MS },
  {
    name: "send-digest",
    data: { type: "hourly" },
  },
);

// Cron expression -- daily at 3:15 AM
await emailQueue.upsertJobScheduler(
  "nightly-cleanup",
  { pattern: "0 15 3 * * *" },
  {
    name: "cleanup",
    data: { type: "nightly" },
    opts: { removeOnComplete: true },
  },
);
Why good: 
upsertJobScheduler
is idempotent (safe to call on every startup), template defines job name/data/opts inherited by each produced job, cron syntax via 
pattern
Gotcha: 
every
intervals align to the clock, not to when you called 
upsertJobScheduler
. A 2000ms interval fires at 0s, 2s, 4s, etc.
See examples/advanced.md for removing schedulers and listing active schedulers.
使用
upsertJobScheduler
（v5.16.0+）处理重复作业。替代已废弃的
repeat
选项。
typescript
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获取移除调度器和列出活动调度器的方法。

Pattern 7: Rate Limiting

模式7：速率限制

Limit how many jobs a Worker processes per time window. The limiter is global across all Workers on the same queue.
typescript
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 },
});
Why good: Global rate limit (10 workers with this config still process max 10 jobs/second total), named constants for limits
For dynamic rate limiting based on external API responses, use 
worker.rateLimit(duration)
+ 
throw Worker.RateLimitError()
. See examples/advanced.md.
限制Worker在指定时间窗口内处理的作业数量。限制器对同一队列的所有Worker全局生效。
typescript
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。

Pattern 8: QueueEvents for Global Monitoring

模式8：QueueEvents全局监控

QueueEvents uses Redis Streams (not Pub/Sub) so events are reliable even across reconnections. Requires its own dedicated connection.
typescript
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}`);
});
Why good: Monitors all Workers on a queue from a single listener, reliable delivery via Redis Streams, separate connection as required
See examples/advanced.md for progress tracking and waiting for specific job completion.
</patterns>
<performance>
QueueEvents使用Redis Streams（而非Pub/Sub），因此即使重新连接，事件也能可靠传递。需要专用的独立连接。
typescript
import { QueueEvents } from "bullmq";

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

queueEvents.on("completed", ({ jobId, returnvalue }) => {
  console.log(`作业 ${jobId} 完成，返回值：${returnvalue}`);
});

queueEvents.on("failed", ({ jobId, failedReason }) => {
  console.error(`作业 ${jobId} 失败，原因：${failedReason}`);
});
优势： 通过单个监听器监控队列上的所有Worker，通过Redis Streams实现可靠交付，按要求使用独立连接
查看examples/advanced.md获取进度跟踪和等待特定作业完成的方法。
</patterns>
<performance>

Performance Optimization

性能优化

  • Concurrency -- Set 
    concurrency
    on Worker options to process multiple jobs in parallel within a single Worker instance: 
    { concurrency: 10 }
    . Only effective for I/O-bound work. CPU-bound work blocks the event loop and causes stalled jobs -- use sandboxed processors instead.
  • Sandboxed processors -- Pass a file path instead of a function to the Worker constructor to run processors in a separate process (or worker thread with 
    useWorkerThreads: true
    ). Prevents CPU-intensive work from blocking lock renewal.
  • addBulk -- Use 
    queue.addBulk([...])
    to add many jobs in a single Redis round-trip instead of calling 
    queue.add()
    in a loop.
  • removeOnComplete/Fail -- Configure auto-removal to prevent unbounded Redis memory growth. Use 
    { count: N }
    to keep the last N jobs for debugging.
  • Separate connections -- Producers (Queue) can share a connection. Workers and QueueEvents each need their own connection due to blocking Redis commands.
</performance>
<decision_framework>
  • 并发 —— 在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命令。
</performance>
<decision_framework>

Decision Framework

决策框架

When to Use BullMQ

是否使用BullMQ

Do you need background job processing?
|-- NO -> Don't use BullMQ
+-- YES -> Do you need persistence, retries, or scheduling?
    |-- NO -> Simple in-process queue or setTimeout may suffice
    +-- YES -> Do you need parent-child job dependencies?
        |-- YES -> BullMQ with FlowProducer
        +-- NO -> Do you need rate limiting or priority?
            |-- YES -> BullMQ with limiter/priority options
            +-- NO -> BullMQ with basic Queue + Worker
是否需要后台作业处理？
|-- 否 -> 不要使用BullMQ
+-- 是 -> 是否需要持久化、重试或调度？
    |-- 否 -> 简单的进程内队列或setTimeout可能足够
    +-- 是 -> 是否需要父子作业依赖？
        |-- 是 -> 使用带FlowProducer的BullMQ
        +-- 否 -> 是否需要速率限制或优先级？
            |-- 是 -> 使用带限制器/优先级选项的BullMQ
            +-- 否 -> 使用基础的Queue + Worker版BullMQ

Which Job Pattern?

选择哪种作业模式？

What kind of job scheduling do you need?
|-- One-time delayed job -> queue.add() with delay option
|-- Recurring on fixed interval -> upsertJobScheduler with every
|-- Recurring on cron schedule -> upsertJobScheduler with pattern
|-- Job that depends on other jobs -> FlowProducer with children
|-- Bulk of independent jobs -> queue.addBulk([...])
需要哪种作业调度？
|-- 一次性延迟作业 -> 使用带delay选项的queue.add()
|-- 固定间隔重复作业 -> 使用带every的upsertJobScheduler
|-- Cron调度重复作业 -> 使用带pattern的upsertJobScheduler
|-- 依赖其他作业的作业 -> 使用带children的FlowProducer
|-- 批量独立作业 -> 使用queue.addBulk([...])

Concurrency Strategy

并发策略

Is the processor CPU-intensive?
|-- YES -> Use sandboxed processor (file path or useWorkerThreads)
+-- NO -> Is it I/O-bound (network calls, DB queries)?
    |-- YES -> Set concurrency option (e.g., 10-50)
    +-- NO -> Default concurrency (1) is fine
</decision_framework>
<red_flags>
处理器是否为CPU密集型？
|-- 是 -> 使用沙箱处理器（文件路径或useWorkerThreads）
+-- 否 -> 是否为I/O密集型（网络调用、数据库查询）？
    |-- 是 -> 设置concurrency选项（如10-50）
    +-- 否 -> 默认并发数（1）即可
</decision_framework>
<red_flags>

RED FLAGS

风险警示

High Priority Issues:
  • Missing 
    connection
    on Queue/Worker/QueueEvents constructor -- BullMQ v5 throws at startup without it
  • Missing 
    maxRetriesPerRequest: null
    on Worker connections -- BullMQ throws immediately
  • No graceful shutdown handler -- in-progress jobs become stalled on process exit and are re-processed by other Workers
  • Using deprecated 
    repeat
    option on 
    queue.add()
    instead of 
    upsertJobScheduler
    -- deprecated since v5.16.0
  • Using integer job IDs -- BullMQ v5 throws; IDs must be strings
Medium Priority Issues:
  • No 
    removeOnComplete
    /
    removeOnFail
    configured -- completed/failed jobs accumulate in Redis indefinitely
  • Sharing a single ioredis connection between Worker and QueueEvents -- both use blocking commands and will interfere
  • CPU-intensive processor without sandboxed mode -- blocks event loop, causes stalled jobs, prevents lock renewal
  • Missing error event handler on Worker -- 
    worker.on("error", ...)
    prevents unhandled errors from crashing the process
Common Mistakes:
  • Assuming 
    worker.close()
    has a timeout -- it waits indefinitely for processors to finish; wrap with your own timeout
  • Using QueueScheduler class -- removed in BullMQ v4; its responsibilities are handled by Workers automatically
  • Expecting 
    limiter
    to be per-Worker -- the rate limit is global across all Workers on the same queue
  • Not making processors idempotent -- BullMQ guarantees at-least-once delivery; a stalled job may be processed twice
Gotchas & Edge Cases:
  • upsertJobScheduler
    every
    intervals align to the clock (0s, 2s, 4s), not to when you called the method
  • worker.close()
    does not cancel running processors -- it waits for them to finish naturally
  • Job priority has a performance cost -- BullMQ uses a different data structure for priority queues; skip if not needed
  • QueueEvents
    uses Redis Streams internally -- ensure your Redis instance has sufficient memory for stream data
  • FlowProducer adds the entire tree atomically -- if any child fails validation, none are added
  • job.getChildrenValues()
    returns an object keyed by 
    "queueName:jobId"
    -- not an array
  • Redis must have 
    maxmemory-policy
    set to 
    noeviction
    -- BullMQ relies on keys not being evicted
</red_flags>
<critical_reminders>
高优先级问题：
  • Queue/Worker/QueueEvents构造函数缺少
    connection
    ——BullMQ v5启动时会抛出错误
  • Worker连接缺少
    maxRetriesPerRequest: null
    ——BullMQ会立即抛出错误
  • 无优雅关闭处理器——进程退出时正在处理的作业会停滞，并被其他Worker重新处理
  • 使用已废弃的
    queue.add()
    中的
    repeat
    选项而非
    upsertJobScheduler
    ——v5.16.0版本起已废弃
  • 使用整数作业ID——BullMQ v5会抛出错误；ID必须为字符串
中优先级问题：
  • 未配置
    removeOnComplete
    /
    removeOnFail
    ——完成/失败的作业会在Redis中无限累积
  • Worker和QueueEvents共享单个ioredis连接——两者都使用阻塞命令，会互相干扰
  • CPU密集型处理器未使用沙箱模式——阻塞事件循环，导致作业停滞，阻止锁续期
  • Worker缺少错误事件处理器——
    worker.on("error", ...)
    防止未处理错误导致进程崩溃
常见错误：
  • 假设
    worker.close()
    有超时——它会无限等待处理器完成；必要时自行添加超时包装
  • 使用QueueScheduler类——在BullMQ v4中已移除；其职责现在由Worker自动处理
  • 认为
    limiter
    是单Worker限制——速率限制对同一队列的所有Worker全局生效
  • 处理器未实现幂等性——BullMQ保证至少一次交付；停滞的作业可能被处理两次
注意事项与边缘情况：
  • upsertJobScheduler
    的
    every
    间隔与时钟对齐（0s、2s、4s），而非调用方法的时间
  • worker.close()
    不会取消运行中的处理器——它会等待处理器自然完成
  • 作业优先级会产生性能开销——BullMQ对优先级队列使用不同的数据结构；非必要时不要使用
  • QueueEvents
    内部使用Redis Streams——确保Redis实例有足够内存存储流数据
  • FlowProducer原子性添加整个作业树——如果任何子作业验证失败，所有作业都不会被添加
  • job.getChildrenValues()
    返回以
    "queueName:jobId"
    为键的对象——而非数组
  • Redis必须将
    maxmemory-policy
    设置为
    noeviction
    ——BullMQ依赖于键不被驱逐
</red_flags>
<critical_reminders>

CRITICAL REMINDERS

重要提醒

All code must follow project conventions in CLAUDE.md (kebab-case, named exports, import ordering, 
import type
, named constants)
(You MUST pass a 
connection
object to every Queue, Worker, QueueEvents, and FlowProducer constructor -- BullMQ v5 throws if connection is missing)
(You MUST set 
maxRetriesPerRequest: null
on ioredis connections used by Workers -- BullMQ requires infinite retries and throws without this setting)
(You MUST call 
await worker.close()
on SIGTERM/SIGINT for graceful shutdown -- without it, in-progress jobs become stalled)
(You MUST use 
upsertJobScheduler
for repeatable/cron jobs -- the old 
repeat
option on 
queue.add
is deprecated since v5.16.0)
Failure to follow these rules will cause startup crashes, stalled jobs, and unreliable job processing.
</critical_reminders>
所有代码必须遵循CLAUDE.md中的项目规范（短横线命名、命名导出、导入顺序、
import type
、命名常量）
（必须为每个Queue、Worker、QueueEvents和FlowProducer构造函数传入
connection
对象——如果缺少连接，BullMQ v5会抛出错误）
（必须为Worker使用的ioredis连接设置
maxRetriesPerRequest: null
——BullMQ要求无限重试，缺少该设置会抛出错误）
（必须在SIGTERM/SIGINT信号触发时调用
await worker.close()
以优雅关闭——否则，正在处理的作业会陷入停滞状态）
（必须使用
upsertJobScheduler
处理重复/定时作业——v5.16.0版本起，
queue.add
中的旧
repeat
选项已废弃）
不遵循这些规则会导致启动崩溃、作业停滞和不可靠的作业处理。
</critical_reminders>