RabbitMQ

Java 高级 · 消息中间件

RabbitMQ 基于 AMQP 协议，以灵活的"交换机-队列"路由和可靠投递著称，非常适合业务系统间的异步解耦（订单、通知、延迟任务）。本文讲清 AMQP 路由模型、Channel 多路复用、ACK/NACK 可靠投递、镜像/仲裁队列与 TTL+死信的实战用法。

一、AMQP 模型：Producer → Exchange → Queue → Consumer

RabbitMQ 不是"发到 Topic 就完事"，而是经过一个路由层：

• Exchange（交换机）：接收消息，按**路由规则（Binding）**分发到 Queue。自身不存消息（除非没人绑）。
• Binding（绑定）：Exchange 与 Queue 之间的规则，带 routingKey 或参数。
• Queue（队列）：真正存消息、被 Consumer 消费。

四种交换机类型

类型	路由规则	典型场景
Direct	routingKey 精确匹配	点对点、定向分发
Topic	routingKey 模式匹配（* 单段，# 多段）	事件分类（最常用）
Fanout	广播到所有绑定队列，忽略 key	通知、缓存失效广播
Headers	按消息头属性匹配（少用）	复杂属性路由

TIP

Topic 的 # 是"0 或多段"，* 是"恰好一段"。如 order.# 匹配 order.created、order.paid.vip；order.* 只匹配 order.paid（单段）。这是做"按业务域订阅"的利器。

二、Channel 多路复用

RabbitMQ 的连接（Connection）是 TCP 长连接，昂贵；在其上可开多个 Channel（信道）——轻量、逻辑隔离的虚拟连接，所有 AMQP 命令都走 Channel。一个 Connection 上多 Channel 复用同一条 TCP，避免每个线程建一条 TCP。

Connection conn = factory.newConnection();     // 一条 TCP
Channel ch1 = conn.createChannel();             // 信道1
Channel ch2 = conn.createChannel();             // 信道2, 复用同 TCP

注意：Channel 不是线程安全的，每个线程应用独立 Channel（用连接池/Channel 复用策略）。Spring AMQP 已帮你管理。

三、可靠投递：ACK / NACK / Publisher Confirm

3.1 消费端 ACK（防丢）

Consumer 处理完才发 basic.ack，Broker 才删消息；若未 ack 且连接断开，消息重回队列（或进死信）。

@RabbitListener(queues = "order.q")
public void on(Message m, Channel ch) throws IOException {
    long tag = m.getMessageProperties().getDeliveryTag();
    try {
        handle(new String(m.getBody()));           // 业务逻辑
        ch.basicAck(tag, false);                    // 处理成功, 确认删除
    } catch (Exception e) {
        ch.basicNack(tag, false, true);             // 失败, requeue=true 重回队列
        // 注意: 必然失败的消息别 requeue=true, 否则死循环
    }
}

3.2 生产端 Publisher Confirm（防发丢）

客户端用 publisher confirm 机制：消息到达 Exchange 并（若持久化）落盘后，Broker 回 ack；若不可路由或落盘失败回 nack。配合持久化交换机/队列，做到"至少一次"投递。

channel.confirmSelect();                 // 开启 confirm 模式
channel.basicPublish(ex, key, props, body);
if (!channel.waitForConfirms(5000)) {    // 确认到达
    // 重试或落本地
}

四、镜像队列与仲裁队列（高可用）

• 镜像队列（Mirrored Queue，经典）：队列在集群多个节点镜像复制，主节点故障自动切镜像（数据不丢）。但镜像同步有性能开销，且脑裂时可能丢消息。
• 仲裁队列（Quorum Queue，推荐，3.8+）：基于 Raft 的法定人数复制，强一致、更抗脑裂，是镜像队列的继任者。新业务优先用 x-queue-type: quorum。

五、TTL 与死信：延迟与兜底

5.1 死信队列（DLX）

消息变成"死信"的三种情况：被 reject/nack 且 requeue=false、消息 TTL 过期、队列超长。通过 x-dead-letter-exchange 把它转入死信队列，供人工/定时补偿。

@Bean public Queue orderQueue() {
    return QueueBuilder.durable("order.q")
        .withArgument("x-dead-letter-exchange", "order.dlx")       // 死信转发的交换机
        .withArgument("x-dead-letter-routing-key", "order.dead")
        .build();
}

5.2 延迟消息（TTL + 死信）

给消息/队列设 TTL，消息过期后变死信进 DLX，实现"延迟投递"。例如订单 30 分钟未支付自动关单：消息发到 delay.q（TTL=30min，死信=关单队列），30 分钟后自动进关单队列被消费。

更简洁的方案是用官方 rabbitmq_delayed_message_exchange 插件（基于定时，不经死信，支持任意延迟），生产推荐。

// 声明 + 绑定(含死信) 示例
@Bean public TopicExchange orderExchange() { return new TopicExchange("order.ex"); }
@Bean public Binding bind() {
    return BindingBuilder.bind(orderQueue()).to(orderExchange()).with("order.#");
}

WARNING

• 自动 ACK 丢消息：autoAck=true 时消息一发出即被认为消费成功，处理中宕机消息丢失。高可靠务必手动 ack。
• 死循环重试打满队列：basicNack(requeue=true) 对必然失败的消息会无限重回，应进死信或记录后丢弃。
• 消息体过大：RabbitMQ 单消息建议 < 数 MB，大文件走对象存储、消息只存 URL。
• 队列无上限堆积：未设 TTL/上限，消费者长期宕机会撑爆内存/磁盘，需配合惰性队列（lazy）与容量监控。

六、生产落地

• 关键链路：publisher confirm + 持久化交换机/队列（durable）+ 手动 ack → "至少一次"。
• 消费端幂等（去重表/版本号），应对重复投递。
• 死信队列统一兜底；延迟消息用延迟插件更简洁。
• 集群 + 仲裁队列保障高可用；监控队列长度、未 ack 数、磁盘水位。

📚 官方文档与参考资料

• RabbitMQ 官方文档：https://www.rabbitmq.com/docs
• RabbitMQ 教程（含 Java 示例）：https://www.rabbitmq.com/tutorials
• RabbitMQ 发布者确认：https://www.rabbitmq.com/docs/confirms

✏️ 练习与自测

1. 用 Topic 交换机设计"订单创建/支付/取消"三类事件的路由键与绑定规则，使不同消费者按需订阅。
2. 说明手动 ACK 与 publisher confirm 分别在哪一端防止消息丢失（消费端 vs 生产端）。
3. 用死信队列 + TTL 实现"5 秒后重试 3 次仍失败则告警"的延迟重试方案，画出流转。
4. 比较镜像队列与仲裁队列在一致性与脑裂处理上的差异，新业务该选哪个？

本节小结

RabbitMQ 的灵活来自"Exchange + Binding 的可编程路由"（Topic 最常用），可靠来自"publisher confirm + 持久化 + 手动 ack"三段保障；死信/延迟（TTL + DLX 或延迟插件）提供业务弹性与兜底。落地口诀：路由用 Topic、投递靠 confirm+ack、失败走死信、高可用靠仲裁队列。