消息队列系列（二）：RabbitMQ 实战与原理

前言：RabbitMQ 是最广泛使用的开源消息代理之一，实现了高级消息队列协议（AMQP）。它的核心杀手锏是灵活的路由机制和成熟的可靠性保障。

一、快速开始：Docker 安装与管理界面

在深入原理之前，我们先把它跑起来。

1. Docker 安装 (带管理界面)

强烈建议使用带 management 后缀的镜像，这样会自带一个 Web 控制台。

CodeBlock Loading...

5672：应用程序连接的端口 (TCP)。
15672：Web 管理控制台端口 (HTTP)。

2. 核心概念：Virtual Host

概念：类似于 MySQL 的 Database，是一个逻辑隔离的命名空间。
作用：不同的项目（如 OrderSystem, UserSystem）可以使用同一个 RabbitMQ 集群，但使用不同的 Virtual Host（如 /order, /user），互不干扰。

二、核心架构模型：AMQP 的精髓

RabbitMQ 与其他 MQ 最大的区别在于引入了 Exchange（交换机） 的概念。生产者通过 Routing Key 将消息发送给 Exchange，Exchange 再根据规则将消息路由到对应的 Queue。

CodeBlock Loading...

1. Exchange 的四种类型

类型	路由逻辑	场景
Direct	精准匹配：Routing Key 必须与 Binding Key 完全一致。	一对一处理，如指定发给“日志处理”队列。
Fanout	广播：忽略 Routing Key，发给所有绑定到该 Exchange 的 Queue。	群发通知，如用户注册后同时通知积分、邮件系统。
Topic	模糊匹配：支持通配符 `*` (匹配一个单词) 和 `#` (匹配多个单词)。	灵活路由，如 `log.*` 匹配 `log.error`, `log.info`。
Headers	Header匹配：根据消息头的属性进行匹配（较少使用）。	特殊业务属性路由。

三、保证消息不丢失：可靠性投递方案

在金融或订单场景中，消息丢失是不可接受的。RabbitMQ 提供了三道防线：

1. 生产者端：Confirm 机制

确保消息成功到达 Broker。

开启 publisher-confirm-type: correlated。
实现 ConfirmCallback 接口，如果 ack 为 true，说明消息已到达 Exchange。

2. Broker 端：持久化 (Persistence)

确保 Broker 重启后数据还在。

Exchange 持久化：声明时 durable=true。
Queue 持久化：声明时 durable=true。
Message 持久化：发送消息时 deliveryMode=2。

3. 消费者端：手动 Ack (Acknowledgement)

确保消息被成功消费业务逻辑。

自动 Ack (默认)：消息一到消费者内存就标记为消费成功。如果业务逻辑报错，消息就丢了。
手动 Ack：关闭自动 Ack，业务处理成功后调用 channel.basicAck()；失败调用 basicNack() 并设置 requeue=true 让消息重回队列。

四、高级特性实战

1. 流量控制 (QoS / Prefetch)

如果队列里有 10000 条消息，消费者启动时默认会全部推送到消费者内存，可能导致消费者 OOM。

解决方案：配置 prefetch_count。
效果：Broker 每次只推 N 条消息给消费者，消费者 Ack 之后，再推下 N 条。
配置：spring.rabbitmq.listener.simple.prefetch=1 (能者多劳)。

2. 死信队列 (DLX - Dead Letter Exchange)

当消息变成“死信”后，会被重新投递到另一个 Exchange（即 DLX）。 消息变成死信的情况：

消息被拒绝 (basicReject/basicNack) 且 requeue=false。
消息 TTL 过期 (Time To Live)。
队列达到最大长度。

用途：排查异常数据，或实现延迟队列。

3. 延迟队列 (Delay Queue)

RabbitMQ 原生不支持延迟队列（虽然有插件），最经典的做法是 TTL + DLX。

步骤：
1. 创建一个没有消费者的队列 Queue_TTL，设置 TTL（如 30分钟），并绑定死信交换机 Exchange_DLX。
2. 消息发给 Queue_TTL。
3. 30分钟后消息过期，进入 Exchange_DLX。
4. 真正的消费者监听 Queue_Target（绑定了 Exchange_DLX），从而实现延迟消费。

五、 Spring Boot 完整配置与代码

1. application.yml

CodeBlock Loading...

2. Java 配置类

CodeBlock Loading...

3. 生产者发送

CodeBlock Loading...

4. 消费者监听

CodeBlock Loading...

前言：RabbitMQ 是最广泛使用的开源消息代理之一，实现了高级消息队列协议（AMQP）。它的核心杀手锏是灵活的路由机制和成熟的可靠性保障。

一、快速开始：Docker 安装与管理界面

在深入原理之前，我们先把它跑起来。

1. Docker 安装 (带管理界面)

强烈建议使用带 management 后缀的镜像，这样会自带一个 Web 控制台。

docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=admin \
  rabbitmq:3-management

CodeBlock Loading...

5672：应用程序连接的端口 (TCP)。
15672：Web 管理控制台端口 (HTTP)。

2. 核心概念：Virtual Host

概念：类似于 MySQL 的 Database，是一个逻辑隔离的命名空间。
作用：不同的项目（如 OrderSystem, UserSystem）可以使用同一个 RabbitMQ 集群，但使用不同的 Virtual Host（如 /order, /user），互不干扰。

二、核心架构模型：AMQP 的精髓

Mermaid Loading...

CodeBlock Loading...

1. Exchange 的四种类型

类型	路由逻辑	场景
Direct	精准匹配：Routing Key 必须与 Binding Key 完全一致。	一对一处理，如指定发给“日志处理”队列。
Fanout	广播：忽略 Routing Key，发给所有绑定到该 Exchange 的 Queue。	群发通知，如用户注册后同时通知积分、邮件系统。
Topic	模糊匹配：支持通配符 `*` (匹配一个单词) 和 `#` (匹配多个单词)。	灵活路由，如 `log.*` 匹配 `log.error`, `log.info`。
Headers	Header匹配：根据消息头的属性进行匹配（较少使用）。	特殊业务属性路由。

三、保证消息不丢失：可靠性投递方案

在金融或订单场景中，消息丢失是不可接受的。RabbitMQ 提供了三道防线：

1. 生产者端：Confirm 机制

确保消息成功到达 Broker。

开启 publisher-confirm-type: correlated。
实现 ConfirmCallback 接口，如果 ack 为 true，说明消息已到达 Exchange。

2. Broker 端：持久化 (Persistence)

确保 Broker 重启后数据还在。

Exchange 持久化：声明时 durable=true。
Queue 持久化：声明时 durable=true。
Message 持久化：发送消息时 deliveryMode=2。

3. 消费者端：手动 Ack (Acknowledgement)

确保消息被成功消费业务逻辑。

自动 Ack (默认)：消息一到消费者内存就标记为消费成功。如果业务逻辑报错，消息就丢了。
手动 Ack：关闭自动 Ack，业务处理成功后调用 channel.basicAck()；失败调用 basicNack() 并设置 requeue=true 让消息重回队列。

四、高级特性实战

1. 流量控制 (QoS / Prefetch)

如果队列里有 10000 条消息，消费者启动时默认会全部推送到消费者内存，可能导致消费者 OOM。

解决方案：配置 prefetch_count。
效果：Broker 每次只推 N 条消息给消费者，消费者 Ack 之后，再推下 N 条。
配置：spring.rabbitmq.listener.simple.prefetch=1 (能者多劳)。

2. 死信队列 (DLX - Dead Letter Exchange)

当消息变成“死信”后，会被重新投递到另一个 Exchange（即 DLX）。 消息变成死信的情况：

消息被拒绝 (basicReject/basicNack) 且 requeue=false。
消息 TTL 过期 (Time To Live)。
队列达到最大长度。

用途：排查异常数据，或实现延迟队列。

3. 延迟队列 (Delay Queue)

RabbitMQ 原生不支持延迟队列（虽然有插件），最经典的做法是 TTL + DLX。

步骤：
1. 创建一个没有消费者的队列 Queue_TTL，设置 TTL（如 30分钟），并绑定死信交换机 Exchange_DLX。
2. 消息发给 Queue_TTL。
3. 30分钟后消息过期，进入 Exchange_DLX。
4. 真正的消费者监听 Queue_Target（绑定了 Exchange_DLX），从而实现延迟消费。

五、 Spring Boot 完整配置与代码

1. application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: admin
    virtual-host: /
    publisher-confirm-type: correlated # 开启发送确认
    publisher-returns: true            # 开启消息回退 (路由失败时返回)
    listener:
      simple:
        acknowledge-mode: manual       # 开启手动 Ack
        prefetch: 1                    # 每次只拉取一条

CodeBlock Loading...

2. Java 配置类

@Configuration
public class RabbitConfig {

    // 1. 配置消息转换器 (使用 JSON 序列化)
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    // 2. 定义死信交换机和死信队列
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange("dlx.exchange");
    }
    
    @Bean
    public Queue dlxQueue() {
        return new Queue("dlx.queue");
    }
    
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("dlx.routing.key");
    }

    // 3. 定义普通队列，绑定死信
    @Bean
    public Queue myQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", "dlx.exchange");
        args.put("x-dead-letter-routing-key", "dlx.routing.key");
        return new Queue("my.queue", true, false, false, args);
    }
}

CodeBlock Loading...

3. 生产者发送

@Autowired
private RabbitTemplate rabbitTemplate;

public void send() {
    // 必须设置 CorrelationData 才能触发 ConfirmCallback
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    rabbitTemplate.convertAndSend("my.direct.exchange", "my.routing.key", new User("Tom", 18), correlationData);
}

CodeBlock Loading...

4. 消费者监听

@Component
public class MyConsumer {

    @RabbitListener(queues = "my.queue")
    public void receive(User user, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            System.out.println("收到消息: " + user);
            // 业务逻辑处理...
            
            // 手动 Ack: 第二个参数 multiple=false 代表只确认当前这条
            channel.basicAck(tag, false);
        } catch (Exception e) {
            try {
                // 出现异常，拒绝消息
                // 第三个参数 requeue=false: 不重回队列，直接进入死信队列 (DLX)
                // 如果是网络抖动等临时异常，可以设为 true 让其重试
                channel.basicNack(tag, false, false);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }
}

CodeBlock Loading...

消息队列系列（二）：RabbitMQ 实战与原理

一、 快速开始：Docker 安装与管理界面

1. Docker 安装 (带管理界面)

2. 核心概念：Virtual Host

二、 核心架构模型：AMQP 的精髓

1. Exchange 的四种类型

三、 保证消息不丢失：可靠性投递方案

1. 生产者端：Confirm 机制

2. Broker 端：持久化 (Persistence)

3. 消费者端：手动 Ack (Acknowledgement)

四、 高级特性实战

1. 流量控制 (QoS / Prefetch)

2. 死信队列 (DLX - Dead Letter Exchange)

3. 延迟队列 (Delay Queue)

五、 Spring Boot 完整配置与代码

1. application.yml

2. Java 配置类

3. 生产者发送

4. 消费者监听

一、 快速开始：Docker 安装与管理界面

1. Docker 安装 (带管理界面)

2. 核心概念：Virtual Host

二、 核心架构模型：AMQP 的精髓

1. Exchange 的四种类型

三、 保证消息不丢失：可靠性投递方案

1. 生产者端：Confirm 机制

2. Broker 端：持久化 (Persistence)

3. 消费者端：手动 Ack (Acknowledgement)

四、 高级特性实战

1. 流量控制 (QoS / Prefetch)

2. 死信队列 (DLX - Dead Letter Exchange)

3. 延迟队列 (Delay Queue)

五、 Spring Boot 完整配置与代码

1. application.yml

2. Java 配置类

3. 生产者发送

4. 消费者监听

一、快速开始：Docker 安装与管理界面

二、核心架构模型：AMQP 的精髓

三、保证消息不丢失：可靠性投递方案

四、高级特性实战

一、快速开始：Docker 安装与管理界面

二、核心架构模型：AMQP 的精髓

三、保证消息不丢失：可靠性投递方案

四、高级特性实战