3、Node.js连接RabbitMQ：amqplib库的使用、建立连接与通道、声明队列与交换机

好，咱们进入正题。RabbitMQ 这东西，说白了就是一个消息中转站。你的 Node.js 应用要跟它打交道，最常用的库就是 amqplib。我这些年做过的项目，只要涉及消息队列，十有八九都是用它。今天我就带你一步步把连接、通道、队列和交换机这些基础操作捋清楚。

3.1 安装与引入 amqplib

先装包，这步没啥好说的：

npm install amqplib

装完之后，在你的代码里引入：

const amqp = require('amqplib');

嗯，这里要注意。amqplib 提供了两种使用方式：回调风格和 Promise 风格。我个人强烈建议用 Promise 风格，配合 async/await，代码读起来清爽很多。你想想看，回调嵌套多了，维护起来多头疼。

3.2 建立连接

连接 RabbitMQ 服务器，核心就是调用 amqp.connect()。它接受一个连接字符串，格式大概是这样的：

amqp://username:password@host:port/vhost

举个例子：

const connection = await amqp.connect('amqp://guest:guest@localhost:5672/');

这里有个坑，我踩过。默认的 guest 用户只能在 localhost 登录。如果你要远程连接，记得新建一个用户，或者给 guest 配置远程访问权限。我曾经在测试环境折腾了半天，一直连不上，最后发现是这个问题。

连接成功后，你会得到一个 connection 对象。这个对象代表与 RabbitMQ 服务器的一个 TCP 连接。它很宝贵，一个应用通常只需要一个连接就够了。

3.3 创建通道

有了连接，下一步就是创建通道。通道是啥？你可以把它理解成连接里的一个轻量级会话。大部分实际操作，比如声明队列、发布消息、消费消息，都是在通道上完成的。

const channel = await connection.createChannel();

为什么要有通道？我简单解释一下。一个 TCP 连接可以创建多个通道，每个通道独立工作。这样你就不需要为每个操作都建立一个新的 TCP 连接，节省资源。说白了，就是复用。

我个人习惯，一个业务模块用一个通道。比如订单模块用一个通道，通知模块用另一个通道。这样出了问题，排查起来也方便。

3.4 声明队列

队列是消息的容器。在 RabbitMQ 里，你要用队列，得先声明它。声明队列用 channel.assertQueue()：

await channel.assertQueue('my_queue', {
  durable: true  // 队列持久化，重启后不丢失
});

这个方法有几个关键参数：

参数	说明	默认值
durable	队列是否持久化。持久化后，RabbitMQ 重启队列还在。	true
exclusive	是否排他。排他队列只对当前连接可见，连接断开自动删除。	false
autoDelete	当最后一个消费者取消订阅后，是否自动删除队列。	false

这里有个细节：assertQueue 是幂等的。也就是说，你多次调用它，如果队列已经存在且参数匹配，它不会报错。如果参数不匹配，它会抛异常。所以声明队列时，参数一定要前后一致。

3.5 声明交换机

交换机是消息的路由器。生产者把消息发给交换机，交换机根据规则把消息投递到对应的队列。声明交换机用 channel.assertExchange()：

await channel.assertExchange('my_exchange', 'direct', {
  durable: true
});

第二个参数是交换机类型，RabbitMQ 支持四种：

• direct：直接匹配。消息的路由键（routing key）必须与队列绑定的路由键完全一致。
• topic：主题匹配。支持通配符 *（匹配一个单词）和 #（匹配零个或多个单词）。
• fanout：广播。消息会发送到所有绑定的队列，忽略路由键。
• headers：头匹配。根据消息的 headers 属性进行匹配，用得比较少。

我在项目中用得最多的是 direct 和 topic。比如订单状态变更，用 direct 就够用了。如果是日志系统，需要按级别和模块过滤，topic 更灵活。

3.6 绑定队列到交换机

声明了队列和交换机，还得把它们绑定起来。绑定用 channel.bindQueue()：

await channel.bindQueue('my_queue', 'my_exchange', 'my_routing_key');

第三个参数是路由键。对于 direct 交换机，这个键就是匹配规则。对于 fanout 交换机，这个键会被忽略。

你想想看，如果不绑定，交换机收到消息后不知道该发给谁，消息就丢了。所以绑定这一步不能少。

3.7 完整示例：一条消息的旅程

说了这么多，咱们来个完整的例子。从连接到发送消息，再到消费消息，一条龙走一遍：

const amqp = require('amqplib');

async function start() {
  // 1. 建立连接
  const connection = await amqp.connect('amqp://localhost');
  
  // 2. 创建通道
  const channel = await connection.createChannel();
  
  // 3. 声明交换机
  await channel.assertExchange('order_exchange', 'direct', { durable: true });
  
  // 4. 声明队列
  await channel.assertQueue('order_queue', { durable: true });
  
  // 5. 绑定队列到交换机
  await channel.bindQueue('order_queue', 'order_exchange', 'order.created');
  
  // 6. 发送消息
  const message = JSON.stringify({ orderId: 123, status: 'created' });
  channel.publish('order_exchange', 'order.created', Buffer.from(message));
  
  console.log('消息已发送');
  
  // 7. 消费消息
  channel.consume('order_queue', (msg) => {
    if (msg) {
      const content = JSON.parse(msg.content.toString());
      console.log('收到消息:', content);
      channel.ack(msg);  // 确认消息已处理
    }
  });
}

start().catch(console.error);

这个例子涵盖了今天讲的所有核心操作。你把它跑起来，就能看到消息从生产者到消费者的完整流程。

3.8 避坑指南

最后，分享几个我踩过的坑：

• 连接未关闭：程序退出前，记得调用 connection.close() 和 channel.close()。否则 RabbitMQ 会一直保留连接，占用资源。
• 消息未确认：消费消息后，一定要调用 channel.ack(msg) 确认。否则消息会一直留在队列里，重复消费。
• 交换机类型选错：我曾经把 direct 和 topic 搞混，结果消息路由不到正确的队列。调试了半天才发现是类型不对。

嗯，今天就到这里。这些基础操作你练熟了，后面讲高级特性时才能跟得上。记住，连接、通道、队列、交换机，这四个概念是 RabbitMQ 的基石。把它们吃透了，后面就顺了。