RabbitMQ

RabbitMQ 是一个基于 AMQP 协议的消息中间件。和 RocketMQ 的 Topic 模型不同，RabbitMQ 的核心是 Exchange、Queue、Binding 三个概念：生产者不直接把消息发到队列，而是先发到交换机，交换机再根据绑定规则把消息路由到一个或多个队列。

为什么 RabbitMQ 要多一层 Exchange

如果生产者直接把消息写进队列，生产者就必须知道所有队列名称。业务一变化，生产者代码就要跟着改。Exchange 的作用是把“发送消息”和“消息去哪些队列”解耦：生产者只关心发到哪个交换机和 routing key，队列如何绑定由 RabbitMQ 配置决定。

flowchart TD
    A["没有 Exchange"] --> B["Producer 直接依赖 Queue"]
    B --> C["队列变化会影响生产者"]
    D["使用 Exchange"] --> E["Producer 发到 Exchange"]
    E --> F["Binding 决定进入哪些 Queue"]
    F --> G["生产者和消费者解耦"]

核心模型

flowchart TD
    P["Producer"] --> E["Exchange<br/>负责路由"]
    E -->|Binding Key| Q1["Queue A<br/>保存消息"]
    E -->|Binding Key| Q2["Queue B<br/>保存消息"]
    Q1 --> C1["Consumer A"]
    Q2 --> C2["Consumer B"]

Producer

生产者负责发布消息。发送时通常需要指定：

• exchange：消息发送到哪个交换机。
• routing key：路由键，交换机根据它决定投递到哪些队列。
• message body：消息内容。
• properties：消息属性，例如持久化、过期时间、消息 ID。

Exchange

交换机负责路由消息，本身不存储消息。常见类型：

类型	说明	适合场景
direct	routing key 完全匹配	点对点任务、明确路由
fanout	广播到所有绑定队列	事件广播、缓存刷新
topic	按通配符匹配 routing key	多维度事件订阅
headers	按消息头匹配	特殊路由，较少使用

Queue

队列负责保存消息，消费者从队列中消费。一个队列可以被多个消费者监听，默认情况下同一条消息只会被一个消费者处理。

Binding

Binding 是 Exchange 和 Queue 之间的绑定关系。它决定交换机收到消息后，应该把消息投递到哪些队列。

Exchange 路由流程

flowchart TD
    A[生产者发送消息] --> B[Exchange 接收消息]
    B --> C{交换机类型}
    C -->|direct| D[完全匹配 routing key]
    C -->|fanout| E[投递到所有绑定队列]
    C -->|topic| F[按通配符匹配]
    D --> G[写入匹配队列]
    E --> G
    F --> G
    G --> H[消费者消费]

消息可靠性

RabbitMQ 的可靠性通常从生产、存储、消费三个阶段处理。

生产阶段

生产者需要确认消息是否真的到达 Broker。常见机制：

• publisher confirm：Broker 确认消息已经接收。
• return callback：消息无法路由到队列时通知生产者。
• mandatory：开启后，无法路由的消息会返回给生产者。

存储阶段

消息要避免 Broker 重启后丢失，需要同时满足：

• Exchange 持久化。
• Queue 持久化。
• Message 设置持久化。

只设置队列持久化还不够，如果消息本身不是持久化消息，Broker 重启后仍可能丢失。

消费阶段

消费端建议使用手动 ACK。业务处理成功后再确认，失败时根据情况选择重新入队或拒绝。

flowchart TD
    A[消费者收到消息] --> B[处理业务]
    B --> C{成功?}
    C -->|是| D[basicAck]
    C -->|否, 可重试| E[basicNack requeue=true]
    C -->|否, 不可重试| F[basicReject requeue=false]
    F --> G[死信队列]

适合场景

• 业务系统之间异步解耦。
• 任务队列，例如邮件发送、文件处理、报表生成。
• 复杂路由，例如不同类型订单进入不同消费者。
• 延迟任务，例如订单超时关闭。
• 需要 AMQP 协议生态的企业应用。

学习目录

• 交换机和路由：重点理解 direct、fanout、topic 的差异。
• ACK、重试和死信：重点理解消息可靠消费。
• 延迟队列：学习 TTL + DLX 的延迟任务实现。

如果你已经学习过 RocketMQ，可以结合 消息队列总览 对比两者模型差异。

代码 Demo：Spring AMQP 发送和消费

发送消息：

@Service
public class OrderMessageService {
    private final RabbitTemplate rabbitTemplate;

    public OrderMessageService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void sendOrderCreated(Long orderId) {
        rabbitTemplate.convertAndSend(
            "order.exchange",
            "order.created",
            Map.of("orderId", orderId)
        );
    }
}

消费消息：

@Component
public class OrderCreatedConsumer {

    @RabbitListener(queues = "order.created.queue")
    public void onMessage(Map<String, Object> message) {
        System.out.println("处理订单创建事件：" + message.get("orderId"));
    }
}

这只是最小用法。生产环境还要配置发布确认、手动 ACK、死信队列和消费幂等。

如果可靠性配置不完整会怎样

阶段	少了什么	后果
生产阶段	没有 publisher confirm	生产者以为发成功，但 Broker 可能没收到
路由阶段	没有 return callback / mandatory	routing key 写错时消息可能被丢弃
存储阶段	Exchange、Queue 或 Message 没持久化	Broker 重启后消息或路由结构丢失
消费阶段	自动 ACK	消费者拿到消息后宕机，业务没执行但消息已删除
重试阶段	没有死信队列	异常消息反复重试或静默丢失