Node.js消息队列(MQ)实战：BullMQ与RabbitMQ应用场景与代码示例

前端开发者通常习惯了“请求 -> 响应”的同步思维模式。但在后端架构中，“解耦”与“异步”才是构建高可用、高并发系统的核心思路。

来看一个典型场景：用户注册成功后，系统需要发送一封欢迎邮件。

• 同步写法：

  await saveUser(); await sendEmail();

  如果发邮件的SMTP服务卡了5秒，用户就会在注册页面转圈5秒，体验极差。

• 异步写法（MQ）：

  await saveUser(); producer.add('send-email', user);

  接口只需0.1秒即可返回“注册成功”的响应。邮件发送这个耗时任务被扔到了消息队列里，由后台独立的Worker进程慢慢处理，实现了请求的快速响应。

今天，我们就用Node.js生态中流行的库BullMQ（基于Redis）以及企业级标准RabbitMQ来实现这套异步架构，聊聊它们各自的应用场景。

01. 核心概念：生产者与消费者

消息队列的架构非常像“回转寿司店”：

1. 生产者 (Producer)：后厨的师傅（你的API接口）。只负责把做好的寿司盘子（任务）放到传送带上，动作极快。
2. 队列 (Queue)：传送带（Redis/RabbitMQ）。负责暂存和运输盘子，起到缓冲和解耦的作用。
3. 消费者 (Consumer)：吃寿司的客人（Worker进程）。负责从传送带上把盘子拿下来“吃掉”（执行具体的耗时任务）。

02. 实战方案一：轻量级 BullMQ (Node.js 首选)

如果你的技术栈以Node.js为主，并且已经使用了Redis，那么BullMQ是绝佳选择。它简单、强大，并且自带失败重试、延迟任务等机制。

1. 生产者（API 接口层）

// producer.ts
import { Queue } from 'bullmq';

// 连接 Redis
const emailQueue = new Queue('email-queue', {
  connection: { host: 'localhost', port: 6379 }
});

// 模拟 API 接口
app.post('/register', async (req, res) => {
  const user = req.body;

  // 1. 存数据库 (略)

  // 2. 扔进队列 (瞬间完成)
  await emailQueue.add('welcome-email', {
    email: user.email,
    name: user.name
  }, {
    delay: 5000, // 甚至可以延迟 5 秒再执行发送任务
    attempts: 3   // 失败后自动重试 3 次
  });

  res.json({ success: true, msg: '注册成功，邮件稍后发送' });
});

2. 消费者（Worker 进程）

这是一个独立运行的脚本，不会阻塞你的主API服务。

// worker.ts
import { Worker } from 'bullmq';

const worker = new Worker('email-queue', async job => {
  console.log(`正在处理任务 ${job.id}: 给 ${job.data.email} 发邮件...`);

  // 模拟耗时操作 (如调用SMTP服务)
  await sleep(2000);

  // 模拟随机失败 (用于测试重试机制)
  if (Math.random() < 0.1) throw new Error('SMTP连接超时');

  console.log('邮件发送成功！');
}, {
  connection: { host: 'localhost', port: 6379 },
  concurrency: 5 // 关键：并发控制。同时只处理5个任务，防止下游服务过载
});

03. 实战方案二：企业级 RabbitMQ

如果你的架构是跨语言的（例如由Node.js生产任务，Java或Go来消费），或者对可靠性和功能有极高的要求（如金融级应用），那么RabbitMQ是行业标准答案。我们使用Node.js的amqplib库来操作。

生产者

import amqp from 'amqplib';

async function sendTask() {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  const queue = 'video_transcode';

  // 声明一个持久化队列，确保服务重启后消息不丢失
  await channel.assertQueue(queue, { durable: true });

  const msg = JSON.stringify({ videoId: 10086 });
  channel.sendToQueue(queue, Buffer.from(msg), { persistent: true });

  console.log("视频转码任务已发送");
}

消费者

async function consumeTask() {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  const queue = 'video_transcode';

  // 关键：公平分发。告诉RabbitMQ，等我处理完并确认一个任务后，再给我发下一个
  channel.prefetch(1);

  channel.consume(queue, async (msg) => {
    if (msg !== null) {
      const data = JSON.parse(msg.content.toString());

      try {
        await processVideo(data.videoId); // 执行耗时的转码操作
        channel.ack(msg); // 手动确认：告诉MQ我已成功处理，可以安全删除消息了
      } catch (e) {
        channel.nack(msg); // 拒绝确认：处理失败，消息可重新入队或进入死信队列(DLX)
      }
    }
  });
}

04. 前端交互：任务扔进队列后，如何获知结果？

这是一个经典的全栈面试题。用户点击“导出Excel”后，后端将任务放入队列并立即返回，前端如何才能知道文件何时生成好？有三种常见方案：

1. 轮询 (Polling)
前端拿到返回的jobId后，每隔几秒调用一次/api/job/status?id=123来查询状态。

• 缺点：频繁请求浪费带宽，且不是真正的实时。

2. WebSocket / 服务器推送事件 (SSE)
Worker处理完任务后，通过建立的WebSocket或SSE连接主动通知前端。

• 缺点：需要额外的开发成本来维护长连接。

3. 站内信或邮件通知
最省事的方案。直接告知用户“任务完成后会将结果发送至您的邮箱”，让用户无需在页面等待。这种方式将异步进行到底，用户体验也更友好。

结语

从开发前端页面到构建后端服务，一个关键的思维转变就是从同步走向异步。消息队列(MQ)就像是高并发系统中的“蓄水池”和“缓冲带”，它有效地解耦了服务，保护了数据库和第三方服务不被突发流量冲垮，极大地提升了系统的整体稳定性和扩展性。

掌握BullMQ或RabbitMQ的使用，理解其背后的生产者-消费者模式，是你从Node.js开发者迈向Node.js应用架构师非常关键的一步。希望本文的实战代码和场景分析能为你带来启发。如果你想深入探讨更多后端架构与中间件技术，欢迎在云栈社区与其他开发者交流。