当前位置：首页 > news >正文

RabbitMQ 全面详解（附面试重点）

news 来源：原创 2025/4/30 16:33:11

RabbitMQ 全面详解（附面试重点）

一、RabbitMQ 与其他消息队列对比

特性	RabbitMQ	Kafka	RocketMQ	ActiveMQ
设计定位	企业级消息中间件（传统业务场景）	高吞吐分布式流处理平台（日志、大数据）	金融级高可靠消息中间件	轻量级 JMS 实现（传统企业集成）
协议支持	AMQP、STOMP、MQTT	自定义协议（基于 TCP）	自定义协议、JMS	AMQP、STOMP、OpenWire
吞吐量	中等（万级 QPS）	极高（百万级 QPS）	高（十万级 QPS）	低（万级 QPS）
消息顺序	队列级别顺序	分区级别顺序	队列级别顺序	队列级别顺序
消息持久化	支持（磁盘持久化）	支持（分段日志存储）	支持	支持
事务支持	支持	不支持（但支持幂等）	支持	支持
典型场景	订单处理、支付回调	日志采集、实时流处理	金融交易、事务消息	企业级系统集成

类比解释：

RabbitMQ 像“快递公司”：保证每个包裹（消息）准确送达，适合对可靠性要求高的场景。
Kafka 像“高速公路”：允许大量车辆（数据）高速通过，适合日志流和大数据管道。
RocketMQ 像“银行运钞车”：强调安全与事务，适合金融等高敏感场景。

二、RabbitMQ 核心架构与底层原理

1. 核心组件

组件	作用	类比
Producer	消息生产者，发送消息到 Exchange	快递员（发送包裹）
Exchange	接收消息并根据规则路由到队列（类似邮局分拣中心）	邮局分拣员
Queue	存储消息的缓冲区，等待 Consumer 消费	快递仓库
Binding	定义 Exchange 和 Queue 的绑定规则（路由键匹配）	快递配送路线表
Consumer	消息消费者，从队列获取消息处理	收件人
Channel	复用 TCP 连接的虚拟通道（轻量级，避免频繁创建连接）	高速公路上的多条车道

2. AMQP 协议工作原理

生产者发布消息 → Exchange
Exchange 路由消息 → 根据类型和 Binding 规则投递到 Queue
消费者订阅 Queue → 通过 Channel 拉取或推送消息
消息确认（ACK） → 确保消息可靠消费

底层原理：

Erlang 语言：基于 Erlang 的 OTP 框架，天生支持高并发和分布式。
消息持久化：通过磁盘存储（消息和队列均可持久化）防止数据丢失。
预取机制（Prefetch）：控制消费者每次获取的消息数量，避免过载。

三、Exchange 类型详解

类型	路由规则	场景示例	代码示例
Direct	精确匹配 Routing Key	订单状态更新（按订单ID路由）	`channel.exchangeDeclare("order", "direct")`
Topic	通配符匹配（`*` 匹配一个词，`#` 匹配多个词）	日志分类（如 `logs.error.*`）	`channel.exchangeDeclare("logs", "topic")`
Fanout	广播到所有绑定队列	系统通知（全员广播）	`channel.exchangeDeclare("alerts", "fanout")`
Headers	匹配消息头（键值对）	复杂条件路由（如设备类型过滤）	`channel.exchangeDeclare("devices", "headers")`

类比：

Direct：快递员按详细地址（Routing Key）送货。
Topic：按地区关键词（如“上海.浦东.*”）批量配送。
Fanout：小区广播通知，每家每户都能收到。
Headers：根据包裹上的标签（Headers）选择配送方式。

四、Spring Cloud 集成 RabbitMQ

1. 添加依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2. 配置连接参数（application.yml）

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

3. 创建交换机、队列与绑定

@Configuration
public class RabbitConfig {

    // 创建直连交换机
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange("order.exchange");
    }

    // 创建队列
    @Bean
    public Queue orderQueue() {
        return new Queue("order.queue", true); // 持久化队列
    }

    // 绑定交换机与队列
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue())
            .to(orderExchange())
            .with("order.routingKey");
    }
}

4. 发送消息

@Service
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendOrderMessage(String orderId) {
        rabbitTemplate.convertAndSend(
            "order.exchange", 
            "order.routingKey", 
            "订单创建: " + orderId
        );
    }
}

5. 消费消息

@Component
public class OrderListener {
    @RabbitListener(queues = "order.queue")
    public void processOrder(String message) {
        System.out.println("收到订单消息: " + message);
    }
}