跳至主要內容

RabbitMQ

硅碳合伙人大约 14 分钟消息中间件RabbitMQAMQPExchange

开篇：RabbitMQ 的核心设计哲学

想象你走进一个大型邮局。你把信交给前台（Exchange），前台根据信封上的信息把信分拣到不同的邮箱（Queue），收件人（Consumer）从各自的邮箱取信。如果信上的地址写错了，信可能会被退回或者丢到"无法投递"的角落（死信队列）。

RabbitMQ 就是按照这个"邮局分拣"模型设计的。它实现了 AMQP（Advanced Message Queuing Protocol） 协议，是目前最流行的开源消息中间件之一。它的核心设计哲学是：灵活的消息路由——通过不同类型的 Exchange 和 Binding，你几乎可以实现任何你能想到的消息分发模式。

和 Kafka 侧重"超高吞吐量"不同，RabbitMQ 更擅长复杂路由、消息可靠投递和中小规模场景。如果你的系统需要灵活的消息分发规则、精确的消息确认机制，RabbitMQ 是一个很好的选择。

一、AMQP 模型：邮局里的角色

整体架构

RabbitMQ 的核心架构由以下角色组成：

  • Producer（生产者）——信件的发送者，把消息投递给 Exchange
  • Exchange（交换机）——邮局的分拣台，根据规则把消息路由到一个或多个 Queue
  • Queue（队列）——信箱，存储消息，等待消费者来取
  • Binding（绑定）——分拣规则，定义了 Exchange 和 Queue 之间的对应关系
  • Consumer（消费者）——收件人，从 Queue 中获取并处理消息
  • VHost（虚拟主机）——类似操作系统的命名空间，用来隔离不同应用的 Exchange、Queue 和权限。每个 VHost 拥有自己独立的资源，互不干扰

连接与信道

Producer 和 Consumer 通过 Connection（TCP 连接） 与 RabbitMQ 通信。一个 Connection 内可以开辟多个 Channel（信道）。

为什么要有 Channel？因为 TCP 连接的创建需要三次握手、断开需要四次挥手，每个线程都建一个连接太浪费了。Channel 是轻量级的逻辑连接，多个线程可以共享同一个 Connection，各自使用自己的 Channel，既节省资源又保证了线程隔离。

类比一下：Connection 是一条高速公路，Channel 是路上的不同车道。大家共用一条路，但各走各的车道，互不干扰。

消息流转过程

完整的消息流转是这样的：

  1. Producer 与 RabbitMQ 建立 Connection，在 Connection 里开辟 Channel
  2. Producer 通过 Channel 把消息发送到指定的 Exchange，消息必须携带 Routing Key
  3. Exchange 根据自身类型和 Binding 规则，决定把消息路由到哪个（些）Queue
  4. 消息在 Queue 中等待
  5. Consumer 同样建立 Connection/Channel，监听指定的 Queue
  6. 有消息到达时，Consumer 进行消费，处理完后发送 ACK 确认

二、四种 Exchange 类型

Exchange 是 RabbitMQ 最灵活也最有特色的部分。根据路由规则的不同，RabbitMQ 提供了四种 Exchange 类型。

2.1 Direct Exchange——精准投递

只有消息的 Routing Key 和 Binding 的 Key 完全匹配时，消息才会被路由到对应的 Queue。

适合点对点通信，就像寄信时写了精确的门牌号。一个 Routing Key 对应一个 Queue，清晰明确。

使用示例：

// 声明 Direct Exchange
channel.exchangeDeclare("my-exchange", BuiltinExchangeType.DIRECT);
// 绑定 Queue
channel.queueBind("order-queue", "my-exchange", "order");
// 发送消息，routing_key="order"
channel.basicPublish("my-exchange", "order", null,
    "订单消息".getBytes());

2.2 Fanout Exchange——全部广播

消息会被发送到所有绑定的 Queue，完全无视 Routing Key。

适合广播场景。比如系统配置变更时，需要通知所有服务刷新缓存；或者用户注册成功后，需要同时触发发邮件、发短信、记积分等多个操作。

// 声明 Fanout Exchange
channel.exchangeDeclare("broadcast", BuiltinExchangeType.FANOUT);
// Fanout 绑定时 routing key 为空字符串
channel.queueBind("email-queue", "broadcast", "");
channel.queueBind("sms-queue", "broadcast", "");
// 发送时 routing key 随便填，反正会被忽略
channel.basicPublish("broadcast", "", null,
    "注册成功".getBytes());

2.3 Topic Exchange——模式匹配

使用通配符进行匹配，是最灵活的路由方式：

  • *（星号）匹配一个单词
  • #（井号）匹配零个或多个单词
Routing Key: order.create.success
Binding Key: order.*.success  → 匹配（*匹配create）
Binding Key: order.#          → 匹配（#匹配create.success）
Binding Key: payment.#        → 不匹配

适合需要灵活过滤的场景。比如日志系统中，不同服务按 服务名.日志级别 作为 Routing Key，监控系统只订阅 *.error，就能收到所有服务的错误日志。

2.4 Headers Exchange——按头信息匹配

根据消息的 Headers 属性（而不是 Routing Key）进行匹配。实际使用较少，因为性能不如前三种，而且 Topic Exchange 基本能覆盖大部分场景。

类型对比速查表

Exchange 类型路由规则Routing Key典型场景
Direct精确匹配必须点对点通信
Fanout广播到所有绑定队列忽略配置刷新、事件通知
Topic通配符匹配支持 * 和 #日志系统、灵活过滤
Headers按消息头匹配不用少用

三、消息可靠性：确保信件不丢

消息从 Producer 到 Consumer 的旅程中，有三个环节可能丢失：Producer -> Exchange、Exchange -> Queue、Queue -> Consumer。RabbitMQ 为每个环节都提供了保障机制。

3.1 生产者确认（Confirm 机制）

生产者把消息发给 Exchange 后，怎么知道 Exchange 收到了？靠 Publisher Confirm 机制。

开启 Confirm 后，RabbitMQ 会在消息成功到达 Exchange 后发送 ACK，失败则发送 NACK。Confirm 是异步的，性能影响很小。

// 开启 Confirm 模式
channel.confirmSelect();

// 注册异步回调
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        System.out.println("消息成功到达 Exchange！");
    }
    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        System.out.println("发送失败，尝试重试或补偿！");
    }
});

3.2 消息路由保障（Return 机制）

消息到达 Exchange 后，如果没有匹配的 Binding，消息就会被默默丢弃。通过 Publisher Return 机制，RabbitMQ 会把无法路由的消息返回给生产者。

channel.addReturnListener((replyCode, replyText, exchange,
    routingKey, properties, body) -> {
    System.out.println("消息没有路由到任何 Queue，做补偿处理！");
});

// 发送时设置 mandatory=true 开启 Return
channel.basicPublish(exchange, routingKey, true, null,
    message.getBytes());

Confirm 保障的是 Producer -> Exchange 这一段，Return 保障的是 Exchange -> Queue 这一段。两者配合使用，才能确保消息从生产者安全到达队列。

3.3 持久化：三层防护

消息到了 Queue 后，如果 RabbitMQ 宕机怎么办？默认情况下消息是存在内存中的，重启就没了。需要做三层持久化：

Queue 持久化——声明队列时设置 durable=true：

// 第二个参数 durable=true，表示队列持久化
channel.queueDeclare("my-queue", true, false, false, null);

Exchange 持久化——声明交换机时同样设置 durable=true：

channel.exchangeDeclare("my-exchange", "direct", true);

消息持久化——发送消息时设置 deliveryMode=2：

AMQP.BasicProperties props = new AMQP.BasicProperties()
    .builder()
    .deliveryMode(2) // 2=持久化，1=非持久化（默认）
    .build();
channel.basicPublish("", "my-queue", true, props,
    message.getBytes());

三层都开启，消息才能在 RabbitMQ 重启后保留。但要注意一个关键点：持久化是异步的。消息先写入内存，再由操作系统异步刷到磁盘。如果在刷盘之前宕机，消息依然会丢。想要 100% 不丢，需要配合本地消息表做兜底。

3.4 消费者确认（ACK 机制）

消费者从 Queue 取到消息后，RabbitMQ 不会立即删除消息，而是等待消费者发送确认。

自动确认（autoAck=true）：消息一投递给消费者就从 Queue 删除。如果消费者处理一半崩溃了，消息就丢了。适合对可靠性要求不高的场景。

手动确认（推荐）：消费者处理完后显式发送 ACK，RabbitMQ 才删除消息。如果消费者崩溃，所有未 ACK 的消息（unacked）会重新变为"就绪"状态（ready），等待下次投递。

// 关闭自动确认（第二个参数 autoAck=false）
channel.basicConsume("my-queue", false, consumer);

// 处理成功，手动发送 ACK
channel.basicAck(envelope.getDeliveryTag(), false);

// 处理失败，发送 NACK，消息重回队列
channel.basicNack(envelope.getDeliveryTag(), false, true);

3.5 可靠性全景图

四道防线环环相扣，才能最大限度保证消息不丢。

四、死信队列与延迟队列

4.1 死信队列（DLQ）

什么样的消息会变成"死信"？三种情况：

  1. 消息被消费者拒绝——发送了 NACK 且设置了不重回队列（requeue=false）
  2. 消息过期——超过了设置的 TTL
  3. 队列满了——队列达到最大长度限制，新消息溢出

如果配置了死信交换机（Dead Letter Exchange），死信会被自动路由到死信队列；如果没有配置，死信就会被直接丢弃。

配置方式是在主队列上设置死信参数：

Map<String, Object> args = new HashMap<>();
// 绑定死信交换机
args.put("x-dead-letter-exchange", "dlx-exchange");
// 设置死信路由键
args.put("x-dead-letter-routing-key", "dlx-routing-key");
channel.queueDeclare("main-queue", true, false, false, args);

死信队列的常见用途：

  • 消息处理失败后的补偿处理
  • 分析为什么消息处理失败
  • 实现延迟消费（见下文）

4.2 基于死信的延迟队列

RabbitMQ 原生不支持延迟队列，但可以巧妙地利用死信机制实现。核心思路：

  1. 创建一个"等待队列"，设置消息的 TTL（比如 15 分钟），但不给它设置消费者
  2. 消息在等待队列中过期后变成死信，自动进入死信队列
  3. 消费者监听死信队列，就实现了延迟消费的效果

典型场景：订单下单后 15 分钟未支付自动关闭。

这个方案有一个坑：队头阻塞。RabbitMQ 只检查队首消息是否过期。如果队首消息的 TTL 是 30 分钟，后面一条消息的 TTL 是 5 分钟，后面这条消息即使过期了也出不来，要等队首先过期。

4.3 延迟交换机插件

为了解决队头阻塞问题，RabbitMQ 官方提供了 rabbitmq_delayed_message_exchange 插件（3.6.12+ 支持）。安装后可以创建 x-delayed-message 类型的交换机。

消息不会立即进入队列，而是先存储在 Mnesia 数据库中，到达指定时间后才投递到队列。这样每条消息的延迟时间互不影响，不存在队头阻塞问题。

限制：

  • 最大延迟时间约 49 天（(2^32)-1 毫秒）
  • 不适合大量延迟消息（如百万级），Mnesia 会成为性能瓶颈
  • 消息存储在单节点磁盘副本上，有丢失风险
  • 大量长时间定时器会导致时间漂移

方案对比

维度死信队列方案延迟交换机插件
队头阻塞有无
延迟精度TTL 级别毫秒级
容量限制无特殊限制百万级以上有瓶颈
复杂度需要声明多组队列安装插件即可
可靠性依赖队列持久化单节点存储，有丢失风险

五、高可用部署

5.1 普通集群模式

多个 RabbitMQ 实例组成集群，共享 Exchange 和 Queue 的元数据（名称、属性、Binding 等），但消息数据只存在于创建它的节点上。

消费者如果连接到了一个没有存储消息的节点，该节点会从实际存储消息的节点拉取数据，再转发给消费者。发送消息也是类似的路径。

优点：增加实例即可扩展存储容量和性能。缺点：如果存储消息的节点挂了，那些消息暂时不可用。

5.2 镜像集群模式

队列的元数据和消息数据在集群中所有节点上同步一份。每次写入消息，都需要同步到所有镜像节点。

优点：任何一个节点挂了，其他节点都有完整数据，真正的高可用。
缺点：写入性能有损耗，因为每条消息都要同步到所有镜像节点。集群越大，同步开销越大。

5.3 消费端限流

当消息大量堆积时，如果 RabbitMQ 把所有消息一股脑推给消费者，消费者可能会被打崩甚至 OOM。通过 basicQos 设置限流：

// prefetchCount=10，每次最多推送 10 条未确认消息
channel.basicQos(0, 10, false);

// 必须关闭自动确认，否则限流无效
channel.basicConsume(queueName, false, consumer);

// 消费者处理完一条后手动 ACK
channel.basicAck(deliveryTag, false);

这样消费者手上最多同时持有 10 条未确认消息。处理完一条、发送 ACK 后，RabbitMQ 才会推送下一条。消费者可以按自己的能力消费，不会被大流量冲垮。

六、防止重复消费

RabbitMQ 有确认机制，正常情况下消费者处理完发送 ACK，消息就会被删除。

但以下情况可能导致重复投递：

  • 消费者处理完但 ACK 因网络延迟丢失，RabbitMQ 以为没消费成功就重投了
  • 消费者处理过程中崩溃重启，未 ACK 的消息重新投递

所以消费者端需要做好幂等控制。通用方案："一锁、二判、三更新"：

  1. 在发送消息时生成一个唯一的业务标识（比如订单号、消息 UUID），放到消息体中
  2. 消费者收到消息后，先用分布式锁锁住这个标识
  3. 查数据库/缓存判断是否已经处理过
  4. 如果没处理过，执行业务逻辑并记录处理状态
  5. 如果已处理过，直接 ACK 丢弃

七、常见面试题精选

1. 如何保障消息一定能发送到 RabbitMQ？

两种方案：

Confirm 机制（推荐）：异步回调方式，Publisher Confirm 保障消息到达 Exchange，Publisher Return 保障消息路由到 Queue。性能好。

事务机制：txSelect() 开启事务，txCommit() 提交，txRollback() 回滚。能保证原子性但性能很差（吞吐量降低 2~10 倍），不推荐在生产环境使用。

2. RabbitMQ 怎么保证高可用？

三种部署模式：

  • 单机模式——只用于开发测试
  • 普通集群——共享元数据，消息在单节点。扩容方便但非真正高可用
  • 镜像集群——元数据和消息全量同步。真正的高可用，代价是写入性能下降

3. 死信队列有什么用？

三大用途：

  • 延迟消费——利用 TTL + 死信实现延迟队列
  • 失败补偿——消费失败的消息进入死信队列，后续人工处理或自动重试
  • 队列溢出保护——队列满了之后消息不会直接丢弃，而是进入死信队列保存

4. RabbitMQ 和 Kafka 怎么选？

维度RabbitMQKafka
协议AMQP自定义协议
路由能力强（4 种 Exchange）弱（按 Partition）
吞吐量万级/秒百万级/秒
消息确认精细（per-message）批量（per-offset）
消息保留消费后删除按时间/大小保留
延迟队列支持（死信/插件）不原生支持
开发语言ErlangScala/Java
最佳场景复杂路由、中小规模日志、大数据、流处理

简单来说：规模大、追求吞吐量选 Kafka；路由复杂、功能丰富选 RabbitMQ。

5. 为什么要使用消息队列？

三大核心价值：

  • 解耦——模块间通过消息通信，互不依赖。下游系统接口变了，上游不用改
  • 异步——耗时操作放队列后台处理，主流程立即返回，提升响应速度
  • 削峰填谷——高峰期请求先进队列排队，下游按能力慢慢消费，防止系统被冲垮

小结

RabbitMQ 的核心优势是灵活的消息路由和精细的消息确认。四种 Exchange 类型覆盖了从点对点到广播的各种分发模式，三层持久化 + Confirm/Return + 手动 ACK 构成了完整的可靠性保障链条。

理解 RabbitMQ，就记住那个邮局模型：Exchange 是分拣台（四种分拣规则），Queue 是信箱（有持久化保障），Binding 是分拣规则（把信投到对应信箱），Consumer 是收件人（取信后签收确认）。

和 Kafka 相比，RabbitMQ 不追求极致吞吐量，而是在消息可靠性、路由灵活性和功能丰富性上做到了最佳平衡。如果你的系统规模适中、路由需求复杂、对消息可靠性要求高，RabbitMQ 是一个值得信赖的选择。