Bullmq Job Queue

何时使用

在 Node.js/TypeScript 应用里需要把耗时或可异步的工作交给 Redis 支撑的队列后台执行时使用，典型场景：

• 发邮件、推送、生成报表等后台任务，请求侧「即发即忘」。
• 延迟任务（N 小时后提醒）、定时/重复任务（每天 9 点跑摘要）。
• 任务优先级、限流（保护下游服务）、失败重试。
• 多步任务流（父任务等待多个子任务完成，如下单 -> 校验库存/扣款/通知）。

不该用（应转交）：

• 需要 Serverless / 边缘队列、不想自己管 Redis -> Upstash QStash。
• 复杂工作流、Saga、补偿事务、长流程编排 -> Temporal 类工作流引擎。
• 事件溯源 / CQRS / 事件流 -> 事件驱动架构。
• Redis 集群、内存调优等基础设施 -> Redis 专项；部署、K8s、扩缩容 -> DevOps。

核心原则： 生产者侧即发即忘；务必显式设置 job options（默认值很少合适）；幂等是你的责任（任务可能执行多次）；重试用指数退避防雪崩；失败任务要有归宿（死信/保留策略）；并发从保守起步；job data 只传 ID 不传大对象；处理 SIGTERM 优雅关闭防止孤儿任务。

步骤

1. 建共享连接：用 ioredis 建连接，必须设 maxRetriesPerRequest: null，否则 Redis 抖动时 Worker 会停摆。
2. 建 Queue 并配默认项：attempts + 指数 backoff + removeOnComplete/removeOnFail 保留策略。
3. 建 Worker：设置 concurrency（从小开始）和 limiter 限流；至少挂 failed 和 stalled 事件处理。
4. 入队任务：请求处理器里尽量不 await queue.add（即发即忘，更快响应）；按需设 delay / repeat / priority。
5. 多步依赖：用 FlowProducer 编排父子任务。
6. 优雅关闭：监听 SIGTERM/SIGINT，先 pause 再 close。
7. 接监控：用 Bull Board 可视化队列与任务状态。

指令

npm i bullmq ioredis
npm i @bull-board/api @bull-board/express   # 可选：可视化监控

环境变量：REDIS_URL（如 redis://localhost:6379，托管可选 Upstash / Redis Cloud / ElastiCache / Railway）。

示例

基础队列 + Worker（任何新队列从这里开始）：

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 小时
});

// 重复任务：按 cron 执行，务必带时区
await queue.add('daily-digest', { type: 'summary' }, {
  repeat: { pattern: '0 9 * * *', tz: 'America/New_York' },
});

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

任务流与依赖（父任务等所有子任务完成）：

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 } },
  ],
});

优雅关闭（部署/重启时不丢任务）：

const shutdown = async () => {
  await worker.pause();   // 停止接新任务
  await worker.close();   // 等当前任务跑完
  await queue.close();    // 关闭队列连接
  process.exit(0);
};
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

Bull Board 监控面板：

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());

注意事项

逐条对应源技能的校验规则，按严重度排列：

• [ERROR] Redis 连接缺 maxRetriesPerRequest: null：Queue/Worker 连接必须设它，否则 Redis 连接出问题时 Worker 会停摆、无法重连。
• [WARN] Worker 未挂 stalled 事件：stalled 任务意味着 Worker 崩溃，必须监控。
• [WARN] Worker 未挂 failed 事件：失败任务要记录并告警。
• [WARN] 无优雅关闭：缺 SIGTERM/SIGINT 处理，部署时任务可能成孤儿。
• [WARN] payload 过大：job data 要小，传 ID 不传完整对象，降低 Redis 内存占用。
• [WARN] 有重试但无退避策略：retry 必须配指数退避，避免惊群（thundering herd）。
• [WARN] 重复任务未显式指定时区：否则用服务器本地时间，会随夏令时漂移。
• [INFO] 请求处理器中 await queue.add：建议即发即忘以加快响应。
• [INFO] 任务无超时配置：建议设超时，防止任务卡死。
• [INFO] Worker 并发过高：确认下游能扛住（数据库连接数、API 限流）。

互见

• Redis 基础设施 / 集群 / 内存调优 -> redis 专项技能
• Serverless / 无 Redis 队列 -> Upstash QStash
• 复杂工作流 / Saga / 长流程编排 -> Temporal 类
• 事件溯源 / CQRS / 事件流 -> 事件架构
• 部署 / K8s / 扩缩容 -> DevOps；监控 / 指标 / 告警 -> 性能与可观测性
• 常见组合：邮件队列栈（bullmq + 邮件系统 + redis）、后台处理栈（bullmq + backend + devops）、AI 处理流水线（bullmq + ai-workflow + 性能监控）、定时任务栈（bullmq + backend + redis）

采编自 sickn33/antigravity-awesome-skills（MIT）。