bullmq-specialist

BullMQ 专家

精通 BullMQ，专注于 Redis 支持的任务队列、后台处理，以及 Node.js/TypeScript 应用中的可靠异步执行。

原则

• 在确认请求前，等待任务持久化入队并处理失败；工作进程可以异步完成任务
• 始终显式设置任务选项——默认值很少适合你的使用场景
• 幂等性由你负责——任务可能会运行多次
• 退避策略可防止惊群效应——指数退避优于线性退避
• 死信队列不可或缺——失败的任务需要有归处
• 并发限制可保护下游服务——从保守设置开始
• 任务数据应尽量精简——传递 ID，而非负载
• 优雅关闭可防止任务成为孤儿——妥善处理 SIGTERM

能力

• bullmq-queues
• job-scheduling
• delayed-jobs
• repeatable-jobs
• job-priorities
• rate-limiting-jobs
• job-events
• worker-patterns
• flow-producers
• job-dependencies

范围

• redis-infrastructure -> redis-specialist
• serverless-queues -> upstash-qstash
• workflow-orchestration -> temporal-craftsman
• event-sourcing -> event-architect
• email-delivery -> email-systems

工具

核心

• bullmq
• ioredis

托管

• upstash
• redis-cloud
• elasticache
• railway

监控

• bull-board
• arena
• bullmq-pro

模式

• delayed-jobs
• repeatable-jobs
• job-flows
• rate-limiting
• sandboxed-processors

模式

基本队列设置

配置完善、可用于生产环境的 BullMQ 队列

适用场景：启动任何新的队列实现

import { Queue, Worker, QueueEvents } from 'bullmq'; import IORedis from 'ioredis';

// 所有队列共用的连接 const connection = new IORedis(process.env.REDIS_URL, { maxRetriesPerRequest: null, // BullMQ 必需 enableReadyCheck: false, });

// 使用合理的默认值创建队列 const emailQueue = new Queue('emails', { connection, defaultJobOptions: { attempts: 3, backoff: { type: 'exponential', delay: 1000, }, removeOnComplete: { count: 1000 }, removeOnFail: { count: 5000 }, }, });

// 带并发限制的工作进程 const worker = new Worker('emails', async (job) => { await sendEmail(job.data); }, { connection, concurrency: 5, limiter: { max: 100, duration: 60000, // 每分钟 100 个任务 }, });

// 处理事件 worker.on('failed', (job, err) => { console.error(Job ${job?.id} failed:, err); });

延迟任务和定时任务

在特定时间或延迟后运行的任务

适用场景：安排未来任务、提醒或定时操作

// 延迟任务——延迟后运行一次 await queue.add('reminder', { userId: 123 }, { delay: 24 * 60 * 60 * 1000, // 24 小时 });

// 可重复任务——按计划运行 await queue.add('daily-digest', { type: 'summary' }, { repeat: { pattern: '0 9 * * *', // 每天上午 9 点 tz: 'America/New_York', }, });

// 移除可重复任务 await queue.removeRepeatable('daily-digest', { pattern: '0 9 * * *', tz: 'America/New_York', });

BullMQ 中同级子任务可以并行运行；它们在数组中的顺序不会按先后顺序执行库存、付款和通知。请将有依赖关系的阶段建模为嵌套依赖或状态机，并确保每个外部操作都具备幂等性。以下代码片段并非可直接用于生产环境的完整代码：请补充 sendEmail/app/queue 定义，通过身份验证/授权保护 Bull Board，并对任务负载和错误信息进行脱敏。请谨慎共享可共用的 Redis 连接；工作进程需要将 maxRetriesPerRequest 设为 null，而 HTTP 生产者应尽快失败。关闭时请关闭 QueueEvents/FlowProducer 和由外部管理的连接。请参阅连接和流程。

作业流和依赖关系

通过父子关系处理复杂的多步骤作业

适用场景：作业需要等待其他作业完成

import { FlowProducer } from 'bullmq';

const flowProducer = new FlowProducer({ connection });

// 父作业等待所有子作业完成 await flowProducer.add({ name: 'process-order', queueName: 'orders', data: { orderId: 123 }, children: [ { name: 'validate-inventory', queueName: 'inventory', data: { orderId: 123 }, }, { name: 'charge-payment', queueName: 'payments', data: { orderId: 123 }, }, { name: 'notify-warehouse', queueName: 'notifications', data: { orderId: 123 }, }, ], });

优雅关闭

妥善关闭 Worker，避免丢失作业

适用场景：部署或重启 Worker

const shutdown = async () => { console.log('Shutting down gracefully...');

// 停止接收新作业 await worker.pause();

// 等待当前作业完成（设定超时时间） await worker.close();

// 关闭队列连接 await queue.close();

process.exit(0); };

process.on('SIGTERM', shutdown); process.on('SIGINT', shutdown);

Bull Board 仪表板

以可视化方式监控 BullMQ 队列

适用场景：需要查看队列状态和作业状态

import { createBullBoard } from '@bull-board/api'; import { BullMQAdapter } from '@bull-board/api/bullMQAdapter'; import { ExpressAdapter } from '@bull-board/express';

const serverAdapter = new ExpressAdapter(); serverAdapter.setBasePath('/admin/queues');

createBullBoard({ queues: [ new BullMQAdapter(emailQueue), new BullMQAdapter(orderQueue), ], serverAdapter, });

app.use('/admin/queues', serverAdapter.getRouter());

验证检查

Redis 连接缺少 maxRetriesPerRequest

严重性：错误

BullMQ 要求将 maxRetriesPerRequest 设为 null，以妥善处理重新连接

消息：创建 BullMQ 队列/Worker 时，Redis 连接未设置 maxRetriesPerRequest: null。这会导致 Worker 在 Redis 连接出现问题时停止运行。

未处理停滞作业事件

严重性：警告

Worker 应处理停滞事件，以检测崩溃的 Worker

消息：创建 Worker 时未设置 'stalled' 事件处理程序。停滞的作业可能表示 Worker 已崩溃，应予以监控。

未处理失败作业事件

严重性：警告

Worker 应处理失败事件，以便监控和发送告警

消息：创建 Worker 时未设置 'failed' 事件处理程序。失败的作业应记录并监控。

未设置优雅关闭处理

严重性：警告

Worker 应在收到 SIGTERM/SIGINT 时优雅关闭

消息：Worker 文件未设置优雅关闭处理。部署时作业可能成为孤立作业。

请求处理程序中的 queue.add 未使用 await

严重性：错误

报告已接受请求前，应等待队列插入完成。应限制生产者重试次数和超时时间，避免 Redis 不可用时 HTTP 请求一直挂起；Worker 的重试设置有所不同。

作业负载中的数据可能过大

严重性：警告

作业数据应保持精简——传递 ID，而不是完整对象

消息：作业似乎包含大量内联数据。请传递 ID，而不是完整对象，以降低 Redis 内存占用。

作业未配置超时

严重性：信息

作业应设置超时，避免无限期执行

消息：添加作业时未显式设置超时。建议添加超时，以防作业卡住。

重试未设置退避策略

严重性：警告

重试应使用指数退避，避免请求激增

消息：作业设置了重试次数，但未设置退避策略。请使用指数退避，避免请求激增。

可重复作业未显式指定时区

严重性：警告

可重复作业应指定时区，以避免夏令时问题

消息：可重复作业未显式指定时区。将使用服务器本地时间，该时间可能因夏令时而偏移。

可能的高工作线程并发

严重程度：信息

高并发可能导致下游服务不堪重负

消息：工作线程并发量较高。请确保下游服务能够承受此负载（数据库连接、API 速率限制）。

协作

委派触发条件

• redis infrastructure|redis cluster|memory tuning -> redis-specialist（队列需要 Redis 基础设施）
• serverless queue|edge queue|no redis -> upstash-qstash（需要无需管理 Redis 的队列）
• complex workflow|saga|compensation|long-running -> temporal-craftsman（需要超越简单任务的工作流编排）
• event sourcing|CQRS|event streaming -> event-architect（需要事件驱动架构）
• deploy|kubernetes|scaling|infrastructure -> devops（队列需要基础设施）
• monitor|metrics|alerting|dashboard -> performance-hunter（队列需要监控）

电子邮件队列技术栈

技能：bullmq-specialist、email-systems、redis-specialist

工作流：

1. Email request received (API)
2. Job queued with rate limiting (bullmq-specialist)
3. Worker processes with backoff (bullmq-specialist)
4. Email sent via provider (email-systems)
5. Status tracked in Redis (redis-specialist)

后台处理技术栈

技能：bullmq-specialist、backend、devops

工作流：

1. API receives request (backend)
2. Long task queued for background (bullmq-specialist)
3. Worker processes async (bullmq-specialist)
4. Result stored/notified (backend)
5. Workers scaled per load (devops)

AI 处理流水线

技能：bullmq-specialist、ai-workflow-automation、performance-hunter

工作流：

1. AI task submitted (ai-workflow-automation)
2. Job flow created with dependencies (bullmq-specialist)
3. Workers process stages (bullmq-specialist)
4. Performance monitored (performance-hunter)
5. Results aggregated (ai-workflow-automation)

定时任务技术栈

技能：bullmq-specialist、backend、redis-specialist

工作流：

1. Repeatable jobs defined (bullmq-specialist)
2. Cron patterns with timezone (bullmq-specialist)
3. Jobs execute on schedule (bullmq-specialist)
4. State managed in Redis (redis-specialist)
5. Results handled (backend)

相关技能

适合与以下技能配合使用：redis-specialist、backend、nextjs-app-router、email-systems、ai-workflow-automation、performance-hunter

适用场景

• 用户提及或暗示：bullmq
• 用户提及或暗示：bull queue
• 用户提及或暗示：redis queue
• 用户提及或暗示：background job
• 用户提及或暗示：job queue
• 用户提及或暗示：delayed job
• 用户提及或暗示：repeatable job
• 用户提及或暗示：worker process
• 用户提及或暗示：job scheduling
• 用户提及或暗示：async processing

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为针对特定环境进行验证、测试或专家审查的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请暂停并要求澄清。