bullmq

BullMQ

当工作涉及 BullMQ 队列、工作进程、流程、重试／调度、NestJS BullMQ、PostgreSQL 与 Redis 后端，或 Pro 与 OSS 的选择时，使用此技能。

快照：[email protected]（2026-10-01）。如果已安装的主版本或次版本不同，请从 source-map.md 刷新。

工作流

1. 修改代码前，检查本地 BullMQ 的实际情况：
   • bullmq、可选的 peer 依赖（ioredis、redis、pg、bullmq-otel）、已存在时的 @nestjs/bullmq、Node/Bun 和 TypeScript 的软件包版本。
   • 后端：默认 Redis（connection 选项或客户端）还是 createPostgresBackend / setDefaultBackendFactory。
   • 连接形式：选项对象、共享客户端还是适配器（createNodeRedisClient、createBunRedisClient、createValkeyGlideClient）；Redis 的 prefix 与 Postgres 的 schema。
   • 生产者（Queue、FlowProducer）与消费者（Worker）；进程隔离方式（内联处理，还是沙盒文件／工作线程）。
   • 默认任务选项、调度器、速率限制、去重用法，以及已存在时的 Nest 模块注册。
2. 如果任务依赖最新 API、版本差异、Pro 功能、适配器、Postgres 或 Nest 封装，请刷新当前官方文档。从 source-map.md 开始。
3. 将工作路由到相应的参考资料：
   • 设置、Redis/Postgres 后端、Queue、Worker、QueueEvents：setup-connections.md。
   • 任务生命周期和 JobsOptions：jobs-options.md。
   • 父子流程／FlowProducer：flows.md。
   • 重试、延迟、任务调度器、速率限制、去重：retries-scheduling-limits.md。
   • 运维、v5→v6 升级、测试、NestJS、Pro 与 OSS、多语言：production-ops-integrations.md。
4. 除非用户明确要求迁移，否则保留代码库现有的连接工厂、Nest 模块和部署拓扑。
5. 在最窄且有用的边界验证行为（处理器单元测试），然后验证队列／工作进程集成。

核心判断

• 将数据存储视为共享基础设施。谨慎复用连接选项或工厂。Worker 和 QueueEvents 仍会为阻塞等待创建重复连接；Redis 客户端／适配器必须支持 duplicate()。
• 默认 Redis 路径需要可选的 ioredis peer 依赖。不要假设 bullmq 仍会捆绑 ioredis。ioredis 6.0（默认使用 RESP3，要求 Node 20+）满足 >=5.0.0 的 peer 依赖范围，但 BullMQ 仍基于 ioredis 5.11 进行测试，且有一个未解决的问题报告称，ioredis 6 的 RedisOptions 不能赋值给 ConnectionOptions。除非项目已使用 ioredis 6，且类型检查和集成测试均通过，否则优先使用 bun add bullmq ioredis@5。
• 非 Redis 后端（Postgres 或自定义后端）：将工厂作为最后一个参数传入，并让 TypeScript 推断后端和连接类型；或者在还需要作业类型参数时，使用 withBackend(createPostgresBackend)（6.3.9+）绑定一次。切勿编写不完整的泛型，例如 Queue<Data, Result, string, PostgresQueueBackend>——Queue 的第 4 个位置不是后端。
• 高层类与数据存储无关。不要使用已移除的 Queue#client、Queue#redisVersion、Queue#databaseType、Worker#blockingClient 或 FlowProducer#client。只有作为备用手段时，才通过 queue.getBackend().client 访问 Redis（类型为 RedisQueueBackend）。
• waitUntilReady() 的返回值是 void。不要将其返回值视为 Redis 客户端。使用 await worker.resume()。
• 在 Queue、Worker、QueueEvents 和 FlowProducer 上添加 error 事件监听器。未处理的连接错误会导致 Node 崩溃。
• 将处理器设计为幂等。作业可能重试、因锁丢失而停滞，或运行多次。
• 谨慎设置 attempts 和 backoff。对于有毒／永久性故障，抛出 UnrecoverableError（Job#discard() 已移除）。抛出 Error 对象，而不是字符串。
• 对于父子关系图，优先使用 FlowProducer。Flow 父作业不得设置 deduplication。Flow 作业选项不包含 repeat。未设置 opts.jobId 的节点会获得 UUID。谨慎选择子作业失败选项（failParentOnFailure、ignoreDependencyOnFailure、removeDependencyOnFailure、continueParentOnFailure）。
• 周期性作业只能使用 Job Schedulers：upsertJobScheduler / getJobSchedulers / removeJobScheduler。在 BullMQ 6.x 中，不要恢复使用 Queue.add(..., { repeat })、Repeat、getRepeatableJobs、removeRepeatable、removeRepeatableByKey、debounce、Job#debounceId、debounced 事件、QueueScheduler 或 v3 之前的 groupKey 速率限制。分组速率限制仅适用于 Pro。
• 去重需要稳定的 deduplication.id。根据需求选择简单去重、节流（ttl）、防抖（extend/replace + delay）或 keepLastIfActive。
• 速率限制在所有 Worker 之间按队列全局生效。优先使用 queue.rateLimit（Worker 的 #rateLimit 已弃用）。可选的队列级 setGlobalConcurrency / setGlobalRateLimit 会限制所有 Worker；每个 Worker 的 concurrency/limiter 不能超过这些限制。
• 优先级：在有优先级的作业中，数值越小，优先级越高（1..2097151）；未设置／0 表示无优先级，会在任何有优先级的作业之前处理。队列暂停时，作业仍处于 waiting 状态——getJobCounts() / JobType 中没有 paused 这种作业状态。
• 调整 removeOnComplete / removeOnFail（按保留时间或数量设置保留策略）。已完成／失败的作业集合若不设限，会无限增长。切勿使用 ioredis 的 keyPrefix；使用 BullMQ 的 prefix（在 Cluster 中使用花括号 hash tag）。Postgres 使用 schema，而不是 Redis 前缀。
• 关停时优雅地关闭 Worker（await worker.close()）。在锁有效期间强制终止进程会导致作业停滞和重复处理，除非运维已预期这种情况。
• 优先使用 @nestjs/bullmq，而不是旧版 @nestjs/bull。除非项目已使用 Pro，否则不要推荐仅限 BullMQ Pro 的 API（分组、分组速率限制、Worker 的 job.getBatch()、可观察对象）。
• 让处理器保持精简，避免阻塞事件循环。需要处理 CPU 密集型任务或隔离崩溃时，使用沙箱处理器／useWorkerThreads。作业的 data 是明文——避免放入密钥，或对字段加密。
• Redis 仍是默认且经过实战检验的后端。仅当项目选择单数据库模式时，才使用 PostgreSQL（createPostgresBackend、显式调用 runMigrations）。同一队列不要混用 Redis 和 Postgres。

验证

优先使用仓库自有的命令。对于有意义的 BullMQ 变更，覆盖相关子集：

• 对 Queue/Worker/FlowProducer 选项（包括后端泛型）、作业的 data/returnvalue/ProgressType 类型，以及使用到的 Nest 注入令牌进行类型检查。
• 独立对处理器成功、可重试失败、UnrecoverableError、进度以及 AbortSignal 取消进行单元测试。
• 针对 Redis 或 Postgres 进行集成测试：添加 → 处理 → 完成/失败；断言作业状态转换。添加作业前执行 await worker.waitUntilReady()（返回 void）。
• 对延迟任务/调度器进行短间隔测试；断言重新部署后不会出现重复调度（upsertJobScheduler）。确认不再使用 repeat / getRepeatableJobs。
• 进行速率限制 / 去重测试：断言忽略/替换/延迟行为，以及 deduplicated 事件（而非 debounced）。
• 流测试：子作业先于父作业完成；父作业通过 getChildrenValues 读取子作业值；子作业失败策略符合产品预期；父节点拒绝 deduplication。
• 更改 lockDuration、stalledInterval、maxStalledCount 或长时间运行的处理器时，进行停滞/锁的冒烟测试。
• 进行优雅关闭冒烟测试：SIGTERM/close 路径应完成正在执行的工作，且在停滞时间窗口之外不应出现意外的孤立锁。
• NestJS 模块冒烟测试：在使用 Nest 时，测试队列注册、WorkerHost 处理器以及 FlowProducer 注入。
• Postgres：已应用 runMigrations；启动时未遇到 SchemaVersionMismatchError / SchemaMigrationRequiredError。
• 更改保留策略（removeOnComplete/removeOnFail）或添加高吞吐量队列时，检查 Redis 增长情况。

报告已运行和未运行的检查，以及仍存在的任何 Pro/OSS、Redis 与 Postgres 之间或软件包版本方面的假设。