BullMQ + Redis 任务队列实战：从入门到生产级架构

你有没有遇到过用户上传一张图片，页面转圈 10 秒才返回结果？或者注册邮件发送失败导致用户流失？这些问题的根源都一样——同步处理阻塞了主流程。任务队列（Task Queue）是解决这类问题最成熟的方案，而 BullMQ 凭借其 Redis 原生支持、简洁 API 和强大功能，已经成为 Node.js 生态中最受欢迎的队列库，周下载量超过 50 万次。

🔧 一、为什么需要任务队列？BullMQ 核心架构

同步 vs 异步：一个真实的性能对比

先看一个最常见的场景：用户注册后发送欢迎邮件。

// ❌ 同步处理：用户必须等待邮件发送完成
app.post('/register', async (req, res) => {
  const user = await db.user.create(req.body);
  await emailService.sendWelcome(user.email); // 阻塞 2-5 秒
  res.json({ success: true, userId: user.id });
});

// ✅ 异步队列：立即返回，后台处理
import { Queue } from 'bullmq';

const emailQueue = new Queue('email', { connection: { host: '127.0.0.1', port: 6379 } });

app.post('/register', async (req, res) => {
  const user = await db.user.create(req.body);
  await emailQueue.add('welcome', { email: user.email, name: user.name }); // < 5ms
  res.json({ success: true, userId: user.id });
});

在 100 并发用户注册的压测下，两者的差距非常明显：

⚡ **关键结论：**异步队列不是「优化」，而是架构层面的必要选择。任何耗时超过 200ms 的操作都不应该阻塞 HTTP 请求。

BullMQ 架构三要素

BullMQ 的核心概念非常简单：

• Producer（生产者）：往队列里添加任务
• Queue（队列）：基于 Redis 的任务存储
• Worker（消费者）：从队列取任务并执行

Producer ──add──> Redis Queue ──process──> Worker
                      │
                  (delayed, prioritized, retried)

安装依赖：

# 安装 BullMQ 和 Redis 客户端
npm install bullmq ioredis

💡 **提示：**BullMQ 要求 Redis 5.0+，推荐使用 Redis 7.x 以获得更好的性能和稳定性。

🚀 二、从简单到高级：5 个实战场景

场景 1：延迟任务——订单超时自动取消

电商系统中，用户下单后 30 分钟未支付，订单需要自动取消。这是延迟任务（Delayed Job）最经典的应用。

// producer.js — 下单时创建延迟任务
import { Queue } from 'bullmq';

const orderQueue = new Queue('orders', {
  connection: { host: '127.0.0.1', port: 6379 },
});

async function createOrder(orderData) {
  const order = await db.order.create({
    ...orderData,
    status: 'pending_payment',
  });

  // 30 分钟后自动检查支付状态
  await orderQueue.add('check-payment', 
    { orderId: order.id },
    { delay: 30 * 60 * 1000 } // 30 分钟，单位毫秒
  );

  return order;
}

// worker.js — 消费延迟任务
import { Worker } from 'bullmq';

const orderWorker = new Worker('orders', async (job) => {
  const { orderId } = job.data;

  if (job.name === 'check-payment') {
    const order = await db.order.findById(orderId);
    
    // 订单已支付，跳过
    if (order.status === 'paid') {
      console.log(`订单 ${orderId} 已支付，跳过取消`);
      return { skipped: true };
    }

    // 未支付，自动取消
    await db.order.update(orderId, { status: 'cancelled' });
    await db.inventory.release(orderId); // 释放库存
    console.log(`订单 ${orderId} 超时未支付，已自动取消`);
    return { cancelled: true };
  }
}, { connection: { host: '127.0.0.1', port: 6379 } });

// 监听任务完成事件
orderWorker.on('completed', (job) => {
  console.log(`任务 ${job.id} 完成:`, job.returnvalue);
});

orderWorker.on('failed', (job, err) => {
  console.error(`任务 ${job.id} 失败:`, err.message);
});

📌 **记住：**延迟任务在 Redis 中存储在 Sorted Set 里，到期后才会被移到等待队列。这意味着即使 Worker 离线再上线，延迟任务也不会丢失。

场景 2：重试策略——图片处理的优雅容错

图片压缩、视频转码这类任务经常因为临时故障失败（磁盘满、内存不足、第三方 API 限流）。合理的重试策略能大幅提高成功率。

import { Queue, Worker } from 'bullmq';

const imageQueue = new Queue('image-processing', {
  connection: { host: '127.0.0.1', port: 6379 },
  defaultJobOptions: {
    attempts: 5,                    // 最多重试 5 次
    backoff: {
      type: 'exponential',          // 指数退避
      delay: 2000,                  // 基础延迟 2 秒
    },
    removeOnComplete: { age: 86400 }, // 完成 24 小时后自动清理
    removeOnFail: { age: 604800 },    // 失败 7 天后自动清理
  },
});

const imageWorker = new Worker('image-processing', async (job) => {
  const { imageUrl, sizes } = job.data;

  // 更新进度
  await job.updateProgress(10);

  const results = [];
  for (const [i, size] of sizes.entries()) {
    const resized = await sharp(imageUrl)
      .resize(size.width, size.height, { fit: 'cover' })
      .webp({ quality: 80 })
      .toBuffer();

    const uploaded = await s3.upload(`images/${job.id}-${size.name}.webp`, resized);
    results.push({ size: size.name, url: uploaded.url });

    // 实时更新进度
    await job.updateProgress(10 + Math.round((i + 1) / sizes.length * 80));
  }

  await job.updateProgress(100);
  return results;
}, {
  connection: { host: '127.0.0.1', port: 6379 },
  concurrency: 3,        // 同时处理 3 个任务
  limiter: {
    max: 10,             // 每秒最多 10 个任务
    duration: 1000,
  },
});

// 失败时记录详细错误
imageWorker.on('failed', (job, err) => {
  console.error(`图片处理失败 (第 ${job.attemptsMade} 次尝试):`, {
    jobId: job.id,
    data: job.data,
    error: err.message,
    stack: err.stack,
  });
});

指数退避的重试间隔如下表所示，能有效避免「惊群效应」：

场景 3：优先级队列——消息推送的 VIP 通道

在消息推送系统中，系统告警比营销推送优先级高得多。BullMQ 支持 1-20 的优先级数值（数值越小优先级越高）。

import { Queue, Worker } from 'bullmq';

const pushQueue = new Queue('push-notifications', {
  connection: { host: '127.0.0.1', port: 6379 },
});

// 系统告警 — 最高优先级
await pushQueue.add('system-alert', 
  { userId: 'admin-001', message: '服务器 CPU 使用率超过 90%' },
  { priority: 1 }
);

// 订单通知 — 高优先级
await pushQueue.add('order-update',
  { userId: 'user-123', message: '您的订单已发货' },
  { priority: 5 }
);

// 营销推送 — 低优先级
await pushQueue.add('marketing',
  { userId: 'user-456', message: '618 大促即将开始！' },
  { priority: 15 }
);

// Worker 处理
const pushWorker = new Worker('push-notifications', async (job) => {
  const { userId, message } = job.data;
  
  switch (job.name) {
    case 'system-alert':
      return await sendUrgentPush(userId, message);
    case 'order-update':
      return await sendNormalPush(userId, message);
    case 'marketing':
      return await sendBatchPush(userId, message);
  }
}, {
  connection: { host: '127.0.0.1', port: 6379 },
  concurrency: 10,
});

⚠️ **警告：**优先级队列会轻微降低吞吐量（约 5-10%），因为 Redis 需要维护优先级排序。仅在真正需要区分优先级时使用，不要所有队列都加优先级。

场景 4：任务依赖链——数据处理流水线

很多业务场景需要多个步骤按顺序执行，比如「数据导入 → 数据清洗 → 生成报表 → 发送通知」。BullMQ 的 FlowProducer 可以优雅地处理这种依赖关系。

import { FlowProducer, Worker } from 'bullmq';

const flow = new FlowProducer({
  connection: { host: '127.0.0.1', port: 6379 },
});

// 定义任务流水线
await flow.add({
  name: 'generate-report',
  queueName: 'reports',
  data: { reportId: 'rpt-001', date: '2026-05-30' },
  children: [
    {
      name: 'clean-data',
      queueName: 'data-processing',
      data: { source: 'raw-data.csv' },
      children: [
        {
          name: 'import-data',
          queueName: 'data-processing',
          data: { fileUrl: 's3://bucket/raw-data.csv' },
        },
      ],
    },
  ],
});

// 每个步骤独立的 Worker
const importWorker = new Worker('data-processing', async (job) => {
  if (job.name === 'import-data') {
    const data = await downloadAndParse(job.data.fileUrl);
    await db.rawData.bulkInsert(data);
    return { rowsImported: data.length };
  }
  if (job.name === 'clean-data') {
    const cleaned = await cleanAndTransform(job.data.source);
    await db.cleanData.bulkInsert(cleaned);
    return { rowsCleaned: cleaned.length };
  }
}, { connection: { host: '127.0.0.1', port: 6379 } });

const reportWorker = new Worker('reports', async (job) => {
  if (job.name === 'generate-report') {
    const pdf = await generatePDF(job.data);
    await emailService.send(job.data.reportId, pdf);
    return { reportUrl: pdf.url };
  }
}, { connection: { host: '127.0.0.1', port: 6379 } });

场景 5：定时任务（Repeatable Jobs）——日报自动生成

BullMQ 内置的 repeat 选项可以替代大部分 node-cron 的使用场景，且天然支持分布式（多个实例不会重复执行）。

import { Queue } from 'bullmq';

const schedulerQueue = new Queue('scheduler', {
  connection: { host: '127.0.0.1', port: 6379 },
});

// 每天早上 9 点生成日报（中国时区）
await schedulerQueue.add('daily-report', 
  { type: 'daily' },
  {
    repeat: {
      pattern: '0 9 * * *',  // cron 表达式
      tz: 'Asia/Shanghai',    // 时区
    },
  }
);

// 每小时清理过期会话
await schedulerQueue.add('cleanup-sessions',
  {},
  {
    repeat: {
      every: 60 * 60 * 1000, // 每小时（毫秒）
    },
  }
);

// 查看所有定时任务
const repeatableJobs = await schedulerQueue.getRepeatableJobs();
console.table(repeatableJobs);
// ┌─────────┬──────────────────┬───────────┬─────────────────┬────────────┐
// │ (index) │       key        │   name    │     pattern     │    next    │
// ├─────────┼──────────────────┼───────────┼─────────────────┼────────────┤
// │    0    │ daily-report:... │daily-report│  0 9 * * *     │ 1748566800 │
// │    1    │ cleanup-sessions │  ...      │ every 3600000  │ 1748563200 │
// └─────────┴──────────────────┴───────────┴─────────────────┴────────────┘

💡 **提示：**BullMQ 的定时任务基于 Redis 的 Sorted Set，天然支持集群部署。10 个实例同时启动，也只会执行一次任务，不需要额外的分布式锁。

📊 三、队列方案选型对比与生产注意事项

BullMQ vs Redis Streams vs RabbitMQ

这三种方案我都在生产环境用过，以下是真实体验的对比：

⚡ **关键结论：**如果团队已经在用 Redis，BullMQ 是性价比最高的选择。它开箱即用的功能（延迟任务、重试、优先级、定时任务）能省去大量造轮子的时间。只有在吞吐量要求超过 50K/s 或需要复杂路由（Topic/Headers Exchange）时，才考虑 RabbitMQ。

生产环境必须注意的 5 个坑

坑点 1：Redis 内存爆炸

任务的 result 和 failedReason 默认会永久存储。如果每天处理 10 万个任务，一个月后 Redis 里会有 300 万条历史记录。

// ✅ 正确做法：设置自动清理策略
const queue = new Queue('my-queue', {
  connection: { host: '127.0.0.1', port: 6379 },
  defaultJobOptions: {
    removeOnComplete: {
      age: 3600,    // 完成 1 小时后删除
      count: 1000,  // 最多保留 1000 条
    },
    removeOnFail: {
      age: 86400 * 7, // 失败 7 天后删除
      count: 5000,
    },
  },
});

坑点 2：Worker 崩溃导致任务卡住

Worker 进程意外退出时，正在执行的任务会变成「active」状态永远不结束。需要配置 lockDuration。

const worker = new Worker('my-queue', processFn, {
  connection: { host: '127.0.0.1', port: 6379 },
  lockDuration: 30000,          // 任务锁 30 秒
  lockRenewTime: 15000,         // 每 15 秒续锁
  stalledInterval: 10000,       // 每 10 秒检查卡住的任务
  maxStalledCount: 2,           // 最多允许卡住 2 次
});

坑点 3：并发控制不当压垮下游服务

Worker 的 concurrency 不是越大越好。如果你的任务会调用第三方 API，高并发会导致限流。

const worker = new Worker('api-calls', processFn, {
  connection: { host: '127.0.0.1', port: 6379 },
  concurrency: 5,  // 同时处理 5 个任务
  limiter: {
    max: 20,       // 每秒最多 20 个任务
    duration: 1000,
    groupKey: 'companyId', // 按公司限流，避免单个客户耗尽配额
  },
});

坑点 4：序列化问题

BullMQ 使用 JSON.stringify 序列化任务数据。Date 对象、Map、Set、undefined 都会丢失。

// ❌ 错误：Date 会被序列化为字符串
await queue.add('task', { createdAt: new Date() }); // Worker 收到的是 "2026-05-30T..."

// ✅ 正确：使用时间戳
await queue.add('task', { createdAt: Date.now() }); // Worker 收到的是数字

坑点 5：Graceful Shutdown

进程退出时不等待任务完成，会导致任务重复执行。

import { Worker } from 'bullmq';

const worker = new Worker('my-queue', processFn, {
  connection: { host: '127.0.0.1', port: 6379 },
});

// 优雅关闭
async function gracefulShutdown(signal) {
  console.log(`收到 ${signal}，正在关闭 Worker...`);
  await worker.close(); // 等待当前任务完成
  process.exit(0);
}

process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
process.on('SIGINT', () => gracefulShutdown('SIGINT'));

监控：Bull Board 可视化面板

开发环境推荐使用 Bull Board 提供 Web 界面监控队列状态：

npm install @bull-board/api @bull-board/express

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

const emailQueue = new Queue('email', { connection: { host: '127.0.0.1', port: 6379 } });
const imageQueue = new Queue('image-processing', { connection: { host: '127.0.0.1', port: 6379 } });

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

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

const app = express();
app.use('/admin/queues', serverAdapter.getRouter());
app.listen(3001, () => console.log('Bull Board: http://localhost:3001/admin/queues'));

✅ 总结与选型建议

BullMQ 不是银弹，但它覆盖了 90% 的后台任务场景。以下是明确的选型建议：

• ✅ 用 BullMQ：Node.js 项目、已有 Redis、需要延迟任务/重试/优先级、团队规模中等
• ✅ 用 Redis Streams：超低延迟场景（< 1ms）、事件溯源、日志聚合
• ✅ 用 RabbitMQ：复杂路由需求、多语言微服务、吞吐量 > 50K/s
• ❌ 别用：简单的 setTimeout 或 setInterval 替代队列——它们没有持久化、没有重试、进程重启就丢失

相关工具推荐：

• Bull Board — 队列监控 Web 面板
• BullMQ Pro — 付费版，支持 Redis Cluster 和更高级的特性
• jsjson.com JSON 格式化工具 — 调试队列数据时格式化 JSON
• jsjson.com 时间戳转换工具 — 调试延迟任务的时间计算