丁丁书屋

RabbitMQ 概述

RabbitMQ 简介

RabbitMQ 是一款基于 AMQP、由 Erlang 语言开发的消息队列产品，2007 年 Rabbit 技术公司发布了它的 1.0 版本

RabbitMQ 体系结构

建立 TCP 连接需要三次握手，反复确认。

所以如果每一次访问 RabbitMQ 服务器都建立一个 Connection 开销会极大，效率低下。所以 Channel 就是在一个已经建立的 Connection 中建立的逻辑连接。如果应用程序支持多线程，那么每个线程创建一个单独的 Channel 进行通讯。每个 Channel 都有自己的 id，Channel 之间是完全隔离的。

总之核心点就是：实现 Connection 复用

RabbitMQ 安装

Docker 安装

bash

# 拉取镜像
docker pull rabbitmq:4.2.0-management

# -d 参数：后台运行 Docker 容器
# --name 参数：设置容器名称
# -p 参数：映射端口号，格式是“宿主机端口号:容器内端口号”。5672供客户端程序访问，15672供后台管理界面访问
# -v 参数：卷映射目录
# -e 参数：设置容器内的环境变量，这里我们设置了登录RabbitMQ管理后台的默认用户和密码
docker run -d \
--name rabbitmq \
-p 35672:5672 \
-p 45672:15672 \
-v ./data:/var/lib/rabbitmq \
-v- ./custom_plugins:/custom_plugins \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=123456 \
-e RABBITMQ_PLUGINS_DIR="/custom_plugins:/plugins"
rabbitmq:4.2.0-management

Docker Compose

yaml

services:
  rabbitmq:
    image: rabbitmq:4.2.0-management
    container_name: rabbitmq
    hostname: rabbitmq
    ports:
      - "35672:5672"
      - "45672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: 123456
      # 冒号分隔多个目录，优先级从左到右
      RABBITMQ_PLUGINS_DIR: "/custom_plugins:/plugins"
    volumes:
      - ./data:/var/lib/rabbitmq
      - ./custom_plugins:/custom_plugins
    restart: unless-stopped

访问后台并登录

RabbitMQ 入门

生产者发送消息，消费者接收消息，用最简单的方式实现

生产者

java

public class Producer {  
  
    public static void main(String[] args) throws Exception {  
  
        // 创建连接工厂  
        ConnectionFactory connectionFactory = new ConnectionFactory();  
  
        // 设置主机地址  
        connectionFactory.setHost("192.168.1.117");
  
        // 设置连接端口号：默认为 5672
        connectionFactory.setPort(35672);
  
        // 虚拟主机名称：默认为 /
        connectionFactory.setVirtualHost("/");
  
        // 设置连接用户名；默认为guest  
        connectionFactory.setUsername("admin");
  
        // 设置连接密码；默认为guest  
        connectionFactory.setPassword("123456");
  
        // 创建连接  
        Connection connection = connectionFactory.newConnection();  
  
        // 创建频道  
        Channel channel = connection.createChannel();  
  
        // 声明（创建）队列  
        // queue      参数1：队列名称  
        // durable    参数2：是否定义持久化队列，当 MQ 重启之后还在  
        // exclusive  参数3：是否独占本次连接。若独占，只能有一个消费者监听这个队列且 Connection 关闭时删除这个队列  
        // autoDelete 参数4：是否在不使用的时候自动删除队列，也就是在没有Consumer时自动删除  
        // arguments  参数5：队列其它参数  
        channel.queueDeclare("simple_queue", true, false, false, null);  
  
        // 要发送的信息  
        String message = "你好；小兔子！";  
  
        // 参数1：交换机名称,如果没有指定则使用默认Default Exchange  
        // 参数2：路由key,简单模式可以传递队列名称  
        // 参数3：配置信息  
        // 参数4：消息内容  
        channel.basicPublish("", "simple_queue", null, message.getBytes());  
  
        System.out.println("已发送消息：" + message);  
  
        // 关闭资源  
        channel.close();  
        connection.close();  
  
    }  
  
}

消费者

java

public class Consumer {
  
    public static void main(String[] args) throws Exception {  
  
        // 1.创建连接工厂  
        ConnectionFactory factory = new ConnectionFactory();  
  
        // 2. 设置参数  
        factory.setHost("192.168.1.117");
        factory.setPort(35672);
        factory.setVirtualHost("/");  
        factory.setUsername("admin");
        factory.setPassword("123456");  
  
        // 3. 创建连接 Connection        
        Connection connection = factory.newConnection();  
  
        // 4. 创建Channel  
        Channel channel = connection.createChannel();  
  
        // 5. 创建队列  
        // 如果没有一个名字叫simple_queue的队列，则会创建该队列，如果有则不会创建  
        // 参数1. queue：队列名称  
        // 参数2. durable：是否持久化。如果持久化，则当MQ重启之后还在  
        // 参数3. exclusive：是否独占。  
        // 参数4. autoDelete：是否自动删除。当没有Consumer时，自动删除掉  
        // 参数5. arguments：其它参数。  
        channel.queueDeclare("simple_queue",true,false,false,null);  
  
        // 接收消息  
        DefaultConsumer consumer = new DefaultConsumer(channel){  
  
            // 回调方法,当收到消息后，会自动执行该方法  
            // 参数1. consumerTag：标识  
            // 参数2. envelope：获取一些信息，交换机，路由key...  
            // 参数3. properties：配置信息  
            // 参数4. body：数据  
            @Override  
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  
  
                System.out.println("consumerTag："+consumerTag);  
                System.out.println("Exchange："+envelope.getExchange());  
                System.out.println("RoutingKey："+envelope.getRoutingKey());  
                System.out.println("properties："+properties);  
                System.out.println("body："+new String(body));  
  
            }  
  
        };  
  
        // 参数1. queue：队列名称  
        // 参数2. autoAck：是否自动确认，类似咱们发短信，发送成功会收到一个确认消息  
        // 参数3. callback：回调对象  
        // 消费者类似一个监听程序，主要是用来监听消息  
        channel.basicConsume("simple_queue",true,consumer);  
  
    }  
  
}

RabbitMQ 经典用法

工作队列模式
发布订阅模式
路由模式
主题模式
远程过程调用

Work Queues 工作队列

RabbitMQ 的工作队列模式就是生产者发送多个消息，由多个消费者来竞争，谁抢到算谁的。

多个消费者监听同一个队列，则各消费者之间对同一个消息是竞争的关系。

Work Queues 工作模式适用于任务较重或任务较多的情况，多消费者分摊任务可以提高消息处理的效率。

工具类

java

public class ConnectionUtil {
  
    public static final String HOST_ADDRESS = "192.168.1.117";
    public static final int PORT = 35672;
    public static final String VIRTUAL_HOST = "/";
    public static final String USER_NAME = "admin";
    public static final String PASSWORD = "123456";

    public static Connection getConnection() throws Exception {

        // 定义连接工厂  
        ConnectionFactory factory = new ConnectionFactory();

        // 设置服务地址  
        factory.setHost(HOST_ADDRESS);

        // 端口  
        factory.setPort(PORT);

        //设置账号信息，用户名、密码、vhost  
        factory.setVirtualHost(VIRTUAL_HOST);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);

        // 通过工程获取连接  
        Connection connection = factory.newConnection();

        return connection;
    }  
  
  
  
    public static void main(String[] args) throws Exception {  
  
        Connection con = ConnectionUtil.getConnection();  
  
        // amqp://guest@192.168.200.100:5672/  
        System.out.println(con);  
  
        con.close();  
  
    }  
  
}

运行的时候先启动两个消费端程序，然后再启动生产者端程序。

消费者 1、消费者 2

java

public class Consumer1 {
  
    static final String QUEUE_NAME = "work_queue";  
  
    public static void main(String[] args) throws Exception {  
  
        Connection connection = ConnectionUtil.getConnection();  
  
        Channel channel = connection.createChannel();  
  
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);  
  
        Consumer consumer = new DefaultConsumer(channel){  
  
            @Override  
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  
  
                System.out.println("Consumer1 body："+new String(body));  
  
            }  
  
        };  
  
        channel.basicConsume(QUEUE_NAME,true,consumer);  
  
    }  
  
}

生产者

java

public class Producer {
  
    public static final String QUEUE_NAME = "work_queue";  
  
    public static void main(String[] args) throws Exception {  
  
        Connection connection = ConnectionUtil.getConnection();  
  
        Channel channel = connection.createChannel();  
  
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);  
  
        for (int i = 1; i <= 10; i++) {  
  
            String body = i+"hello rabbitmq~~~";  
  
            channel.basicPublish("",QUEUE_NAME,null,body.getBytes());  
  
        }  
  
        channel.close();  
  
        connection.close();  
  
    }  
  
}

效果

Pub/Sub 发布/订阅模式

引入新角色——交换机

生产者不是把消息直接发送到队列，而是发送到交换机
交换机接收消息，而如何处理消息取决于交换机的类型
交换机有如下 3 种常见类型
- Fanout：广播，将消息发送给所有绑定到交换机的队列，对应发布/订阅模式
- Direct：定向，把消息交给符合指定 routing key 的队列，对应路由模式
- Topic：通配符，把消息交给符合 routing pattern（路由模式）的队列，对应主题模式

注意：Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！

生产者

java

public class Producer {
  
    public static void main(String[] args) throws Exception {  
  
      // 1、获取连接  
        Connection connection = ConnectionUtil.getConnection();  
  
      // 2、创建频道  
        Channel channel = connection.createChannel();  
  
        // 参数1. exchange：交换机名称  
        // 参数2. type：交换机类型  
        //     DIRECT("direct")：定向  
        //     FANOUT("fanout")：扇形（广播），发送消息到每一个与之绑定队列。  
        //     TOPIC("topic")：通配符的方式  
        //     HEADERS("headers")：参数匹配  
        // 参数3. durable：是否持久化  
        // 参数4. autoDelete：自动删除  
        // 参数5. internal：内部使用。一般false  
        // 参数6. arguments：其它参数  
        String exchangeName = "test_fanout";  
  
        // 3、创建交换机  
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);  
  
        // 4、创建队列  
        String queue1Name = "test_fanout_queue1";  
        String queue2Name = "test_fanout_queue2";  
  
        channel.queueDeclare(queue1Name,true,false,false,null);  
        channel.queueDeclare(queue2Name,true,false,false,null);  
  
        // 5、绑定队列和交换机  
      // 参数1. queue：队列名称  
      // 参数2. exchange：交换机名称  
      // 参数3. routingKey：路由键，绑定规则  
      //     如果交换机的类型为fanout，routingKey设置为""  
        channel.queueBind(queue1Name,exchangeName,"");  
        channel.queueBind(queue2Name,exchangeName,"");  
  
        String body = "日志信息：张三调用了findAll方法...日志级别：info...";  
  
        // 6、发送消息  
        channel.basicPublish(exchangeName,"",null,body.getBytes());  
  
        // 7、释放资源  
        channel.close();  
        connection.close();  
  
    }  
  
}

消费者 1 号 2 号

java

public class Consumer1 {
  
    public static void main(String[] args) throws Exception {  
  
        Connection connection = ConnectionUtil.getConnection();  
  
        Channel channel = connection.createChannel();  
  
        String queue1Name = "test_fanout_queue1";  
  
        channel.queueDeclare(queue1Name,true,false,false,null);  
  
        Consumer consumer = new DefaultConsumer(channel){  
  
            @Override  
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  
  
                System.out.println("body："+new String(body));  
                System.out.println("队列 1 消费者 1 将日志信息打印到控制台.....");  
  
            }  
  
        };  
  
        channel.basicConsume(queue1Name,true,consumer);  
  
    }  
  
}

发布订阅模式与工作队列模式的区别：

工作队列模式本质上是绑定默认交换机
发布订阅模式绑定指定交换机
监听同一个队列的消费端程序彼此之间是竞争关系
绑定同一个交换机的多个队列在发布订阅模式下，消息是广播的，每个队列都能接收到消息

Routing 路由模式

通过『路由绑定』的方式，把交换机和队列关联起来
交换机和队列通过路由键进行绑定
生产者发送消息时不仅要指定交换机，还要指定路由键
交换机接收到消息会发送到路由键绑定的队列
在编码上与 Publish/Subscribe 发布与订阅模式的区别：
- 交换机的类型为：Direct
- 队列绑定交换机的时候需要指定 routing key。

生产者

java

public class Producer {  
  
    public static void main(String[] args) throws Exception {  
  
      Connection connection = ConnectionUtil.getConnection();  
  
      Channel channel = connection.createChannel();  
  
      String exchangeName = "test_direct";  
  
      // 创建交换机  
      channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT,true,false,false,null);  
  
      // 创建队列  
      String queue1Name = "test_direct_queue1";  
      String queue2Name = "test_direct_queue2";  
  
      // 声明（创建）队列  
      channel.queueDeclare(queue1Name,true,false,false,null);  
      channel.queueDeclare(queue2Name,true,false,false,null);  
  
      // 队列绑定交换机  
      // 队列1绑定error  
      channel.queueBind(queue1Name,exchangeName,"error");  
  
      // 队列2绑定info error warning  
      channel.queueBind(queue2Name,exchangeName,"info");  
      channel.queueBind(queue2Name,exchangeName,"error");  
      channel.queueBind(queue2Name,exchangeName,"warning");  
  
        String message = "日志信息：张三调用了delete方法.错误了,日志级别warning";  
  
        // 发送消息  
        channel.basicPublish(exchangeName,"warning",null,message.getBytes());  
        System.out.println(message);  
  
      // 释放资源  
        channel.close();  
        connection.close();  
  
    }  
  
}

接收者 1 2

java

public class Consumer1 {
  
    public static void main(String[] args) throws Exception {  
  
        Connection connection = ConnectionUtil.getConnection();  
  
        Channel channel = connection.createChannel();  
  
        String queue1Name = "test_direct_queue1";  
  
        channel.queueDeclare(queue1Name,true,false,false,null);  
  
        Consumer consumer = new DefaultConsumer(channel){  
  
            @Override  
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  
  
                System.out.println("body："+new String(body));  
                System.out.println("Consumer1 将日志信息打印到控制台.....");  
  
            }  
  
        };  
  
        channel.basicConsume(queue1Name,true,consumer);  
  
    }  
  
}

Topic 主题模式

Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型 Exchange 可以让队列在绑定 Routing key 的时候使用通配符
Routingkey 一般都是由一个或多个单词组成，多个单词之间以“.”分割，例如：item.insert
通配符规则：
- #：匹配零个或多个词
- *：匹配一个词

生产者

java

public class Producer {
  
    public static void main(String[] args) throws Exception {  
  
        Connection connection = ConnectionUtil.getConnection();  
  
        Channel channel = connection.createChannel();  
  
        String exchangeName = "test_topic";  
  
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);  
  
        String queue1Name = "test_topic_queue1";  
        String queue2Name = "test_topic_queue2";  
  
        channel.queueDeclare(queue1Name,true,false,false,null);  
        channel.queueDeclare(queue2Name,true,false,false,null);  
  
        // 绑定队列和交换机  
      // 参数1. queue：队列名称  
      // 参数2. exchange：交换机名称  
      // 参数3. routingKey：路由键,绑定规则  
      //      如果交换机的类型为fanout ,routingKey设置为""  
        // routing key 常用格式：系统的名称.日志的级别。  
        // 需求： 所有error级别的日志存入数据库,所有order系统的日志存入数据库  
        channel.queueBind(queue1Name,exchangeName,"#.error");  
        channel.queueBind(queue1Name,exchangeName,"order.*");  
        channel.queueBind(queue2Name,exchangeName,"*.*");  
  
        // 分别发送消息到队列：order.info、goods.info、goods.error  
        String body = "[所在系统：order][日志级别：info][日志内容：订单生成，保存成功]";  
        channel.basicPublish(exchangeName,"order.info",null,body.getBytes());  
  
        body = "[所在系统：goods][日志级别：info][日志内容：商品发布成功]";  
        channel.basicPublish(exchangeName,"goods.info",null,body.getBytes());  
  
        body = "[所在系统：goods][日志级别：error][日志内容：商品发布失败]";  
        channel.basicPublish(exchangeName,"goods.error",null,body.getBytes());  
  
        channel.close();  
        connection.close();  
  
    }  
  
}

消费者 1 2

java

public class Consumer1 {
  
    public static void main(String[] args) throws Exception {  
  
        Connection connection = ConnectionUtil.getConnection();  
  
        Channel channel = connection.createChannel();  
  
        String QUEUE_NAME = "test_topic_queue1";  
  
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);  
  
        Consumer consumer = new DefaultConsumer(channel){  
  
            @Override  
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  
  
                System.out.println("body："+new String(body));  
  
            }  
  
        };  
  
        channel.basicConsume(QUEUE_NAME,true,consumer);  
  
    }  
  
}

RPC 模式

远程过程调用，本质上是同步调用，和我们使用 OpenFeign 调用远程接口一样，不是典型的消息队列工作方式，不展开说明。

RabbitMQ 整合 SpringBoot

引入依赖

xml

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>4.1.0</version>
</parent>
    
<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
    </dependencies>

项目结构

RabbitMQ 配置

yaml

spring:
  rabbitmq:
    host: 192.168.1.117
    port: 35672
    username: admin
    password: 123456
    virtual-host: /

消费者端

日志配置

yaml

logging:
  level:
    com.atguigu.mq.listener.MyMessageListener: info

监听器类接收消息，在接收消息的方法上应用注解

java

@Component
@Slf4j
public class MyMessageListener {
  
    public static final String EXCHANGE_DIRECT = "exchange.direct.order";  
    public static final String ROUTING_KEY = "order";  
    public static final String QUEUE_NAME  = "queue.order";  
  
    // 设定基本信息 队列名称 持久化 指定交换机信息 指定路由键信息
    @RabbitListener(bindings = @QueueBinding(// 设置绑定关系   
            // 配置队列信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
            value = @Queue(value = QUEUE_NAME, durable = "true", autoDelete = "false"),
            // 配置交换机信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
            exchange = @Exchange(value = EXCHANGE_DIRECT, durable = "true", autoDelete = "false"),
            // 配置路由键信息
            key = {ROUTING_KEY}
    ))
    public void processMessage(String dateString,
                               Message message,
                               Channel channel) { // 接收消息的方法 消息数据本身 消息对象 频道对象
        log.info(dateString);
    }
  
}

@RabbitListener

bindings 属性

@RabbitListener(bindings = @QueueBinding(// 设置绑定关系   
            // 配置队列信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
            value = @Queue(value = QUEUE_NAME, durable = "true", autoDelete = "false"),
            // 配置交换机信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
            exchange = @Exchange(value = EXCHANGE_DIRECT, durable = "true", autoDelete = "false"),
            // 配置路由键信息
            key = {ROUTING_KEY}
    ))

指定交换机和队列之间的绑定关系，指定当前方法要监听的队列

如果 RabbitMQ 服务器上没有这里指定的交换机和队列，那么框架底层的代码会创建它们

queues 属性

java

@RabbitListener(queues = {QUEUE_ATGUIGU})

指定当前方法要监听的队列

此时框架不会创建相关交换机和队列，必须提前创建好

生产者端

java

@SpringBootTest
public class RabbitMQTest {  
  
    public static final String EXCHANGE_DIRECT = "exchange.direct.order";  
    public static final String ROUTING_KEY = "order";
  
    @Autowired  
    private RabbitTemplate rabbitTemplate;
  
    @Test  
    public void testSendMessage() {  
        rabbitTemplate.convertAndSend(  
                EXCHANGE_DIRECT,   
                ROUTING_KEY,   
                "Hello word!");
    }  
  
}

RabbitMQ 消息可靠性投递

消息发送的可靠性问题

下单正常流程

故障情况一：消息没有发送到消息队列上。

解决方案：

在 生产者端进行确认，具体操作中我们会分别针对交换机和队列来确认，如果没有成功发送到消息队列服务器上，那就可以尝试重新发送
为目标交换机指定 备份交换机，当目标交换机投递失败时，把消息投递至备份交换机

故障情况 2：消息成功存入消息队列，但是消息队列服务器宕机了，原本保存在内存中的消息也丢失了，即使服务器重新启动，消息也找不回来了。

解决方案：消息 持久化 到硬盘上，哪怕服务器重启也不会导致消息丢失

故障情况 3：消息成功存入消息队列，但是消费端出现问题，例如：宕机、抛异常等等，导致致消息没有成功被消费

解决方案：

消费端消费消息成功，给服务器返回 ACK 信息，然后消息队列删除该消息
消费端消费消息失败，给服务器端返回 NACK 信息，同时把消息恢复为待消费的状态，这样就可以再次取回消息，重试一次（当然，这就需要消费端接口支持幂等性）

生产者消息确认机制

必须要增加的配置

yaml

spring:
  rabbitmq:
    publisher-confirm-type: CORRELATED # 交换机的确认
    publisher-returns: true # 队列的确认

创建配置类

需要声明回调函数来接收 RabbitMQ 服务器返回的确认信息

确认消息是否发送到交换机
确认消息是否发送到队列

对 RabbitTemplate 的功能进行增强，因为回调函数所在对象必须设置到 RabbitTemplate 对象中才能生效。配置类自身实现 RabbitTemplate.ConfirmCallback、RabbitTemplate.ReturnCallback 这两个接口

java

@Component
@Slf4j
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("消息发送到交换机成功！数据：" + correlationData + " 原因：" + cause);
        } else {
            log.info("消息发送到交换机失败！数据：" + correlationData + " 原因：" + cause);
        }
    }

    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.info("消息主体: " + new String(returned.getMessage().getBody()));
        log.info("应答码: " + returned.getReplyCode());
        log.info("描述：" + returned.getReplyText());
        log.info("消息使用的交换器 exchange : " + returned.getExchange());
        log.info("消息使用的路由键 routing : " + returned.getRoutingKey());
    }
}

@PostConstruct 注解是 Java 中的一个标准注解，它用于指定在对象创建之后立即执行的方法。当使用依赖注入（如 Spring 框架）或者其他方式创建对象时，@PostConstruct 注解可以确保在对象完全初始化之后，执行相应的方法。
使用 @PostConstruct 注解的方法必须满足以下条件：
方法不能有任何参数。
方法必须是非静态的。
方法不能返回任何值。
当容器实例化一个带有 @PostConstruct 注解的 Bean 时，它会在调用构造函数之后，并在依赖注入完成之前调用被 @PostConstruct 注解标记的方法。这样，我们可以在该方法中进行一些初始化操作，比如读取配置文件、建立数据库连接等。

测试效果：

正常情况
交换机名不存在
路由键不存在

备份交换机

原理

步骤

添加备份交换机
添加队列
队列绑定交换机
创建一个交换机并指定备份交换机

效果

rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, "noexistskey", "Hello word!");

指定一个不存在的路由键，发送消息，成功发到备用交换机的队列中。

消息持久化

如果是临时性的交换机和队列，重启之后都没了。队列是消息存储的容器，队列没了，消息就也跟着没了。

创建持久化的交换机和队列，实现消息持久化。

@Queue 注解 durable 属性设置为 true，或者设置 queueName 队列默认是持久化的。
@Exchange 注解 durable 属性，默认为 true
在后台管理界面创建交换机和队列时，默认就是持久化的模式。

消费端消息确认机制

ACK 是 acknowledge 的缩写，表示已确认

默认情况下，消费端取回消息后，默认会自动返回 ACK 确认消息，RabbitMQ 得到 ACK 确认信息就会删除消息。但实际开发中，消费端根据消息队列投递的消息执行对应的业务，未必都能执行成功，因此默认设定就不满足要，要修改成手动确认。

关键配置

yaml

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual # 把消息确认模式改为手动确认

deliveryTag 交付标签机制：

消息往消费端投递时，会携带交付标签

消费端把消息处理结果 ACK、NACK、Reject 等返回给 Broker 之后，Broker 需要对对应的消息执行后续操作，例如删除消息、重新排队或标记为死信等等。那么 Broker 就必须知道它现在要操作的消息具体是哪一条。而 deliveryTag 作为消息的唯一标识就很好的满足了这个需求。

deliveryTag 在 Broker 范围内唯一

削峰限流

设定 prefetch

关键配置

yaml

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 设置每次最多从消息队列服务器取回多少消息

生产 100 条消息

java

@Test
    public void testSendMessage2() {
        for (int i = 0; i < 100; i++) {
            rabbitTemplate.convertAndSend(
                    EXCHANGE_DIRECT,
                    ROUTING_KEY ,
                    "Test prefetch!" + i);
        }
    }

消费端监听

java

@RabbitListener(bindings = @QueueBinding(// 设置绑定关系
            // 配置队列信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
            value = @Queue(value = QUEUE_NAME, durable = "true", autoDelete = "false"),
            // 配置交换机信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
            exchange = @Exchange(value = EXCHANGE_DIRECT, durable = "true", autoDelete = "false"),
            // 配置路由键信息
            key = {ROUTING_KEY}
    ))
    public void processMessageTestPrefetch(String dataString,
                               Message message,
                               Channel channel) throws IOException, InterruptedException {
        log.info("消费端接收到消息内容：" + dataString);
        TimeUnit.SECONDS.sleep(1);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

效果：消费端直接取出所有消息，不断处理消息，直至所有消息确认。

配置最多取回多少消息，参数：prefetch

yaml

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 1 # 设置每次最多从消息队列服务器取回多少消息

效果：一次最多只有取一条消息进行处理

消息超时

给消息设定一个过期时间，超过这个时间没有被取走的消息就会被删除
我们可以从两个层面来给消息设定过期时间：
- 队列层面：在队列层面设定消息的过期时间，并不是队列的过期时间。意思是这个队列中的消息全部使用同一个过期时间。
- 消息本身：给具体的某个消息设定过期时间
如果两个层面都做了设置，那么哪个时间短，哪个生效

添加交换机

添加带过期时间的队列 x-message-ttl

用路由键绑定队列

生产消息

java


    public static final String EXCHANGE_TIMEOUT = "exchange.test.timeout";

    public static final String ROUTING_KEY_TIMEOUT = "timeout";
    
	@Test
    public void testSendMessage3() {
        rabbitTemplate.convertAndSend(
                EXCHANGE_TIMEOUT,
                ROUTING_KEY_TIMEOUT ,
                "Test timeout!");
    }

效果 5 分钟后消息自动失效

消息层面设置超时

java

@Test
    public void testSendMessage3() {
        MessagePostProcessor messagePostProcessor = message -> {
            message.getMessageProperties().setExpiration("2000");
            return message;
        };
        rabbitTemplate.convertAndSend(
                EXCHANGE_TIMEOUT,
                ROUTING_KEY_TIMEOUT ,
                "Test timeout!", messagePostProcessor);
    }

死信和死信队列

概念：当一个消息无法被消费，它就变成了死信。
死信产生的原因大致有下面三种：
- 拒绝：消费者拒接消息，basicNack()/basicReject()，并且不把消息重新放入原目标队列，requeue = false
- 溢出：队列中消息数量到达限制。比如队列最大只能存储 10 条消息，且现在已经存储了 10 条，此时如果再发送一条消息进来，根据先进先出原则，队列中最早的消息会变成死信
- 超时：消息到达超时时间未被消费
死信的处理方式大致有下面三种：
- 丢弃：对不重要的消息直接丢弃，不做处理
- 入库：把死信写入数据库，日后处理
- 监听：消息变成死信后进入死信队列，我们专门设置消费端监听死信队列，做后续处理（通常采用）

创建死信交换机和死信队列

死信交换机：exchange.dead.letter.video
死信队列：queue.dead.letter.video
死信路由键：routing.key.dead.letter.video

创建正常交换机和正常队列

正常交换机：exchange.normal.video
正常队列：queue.normal.video
- 指定死信交换机
- 指定死信路由键
- 指定消息容量
- 指定超时时间
正常路由键：routing.key.normal.video

情况一：拒绝

java

	public static final String QUEUE_NORMAL = "queue.normal.video";
    public static final String QUEUE_DEAD_LETTER = "queue.dead.letter.video";


	@RabbitListener(queues = {QUEUE_NORMAL})
    public void processMessageNormal(String dataString,
                                     Message message,
                                     Channel channel) throws IOException {
        log.info("消费端接收到消息内容：" + dataString + "，拒绝");
        channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(queues = {QUEUE_DEAD_LETTER})
    public void processMessageDead(String dataString,
                                     Message message,
                                     Channel channel) throws IOException {
        log.info("监听到死信消息内容：" + dataString);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

情况二：溢出

java

	public static final String EXCHANGE_NORMAL = "exchange.normal.video";

    public static final String ROUTING_KEY_NORMAL = "routing.key.normal.video";

    @Test
    public void testSendMessage4() {

        for (int i = 0; i < 20; i++) {
            rabbitTemplate.convertAndSend(
                    EXCHANGE_NORMAL,
                    ROUTING_KEY_NORMAL ,
                    "Test max length!" + i);
        }
    }

情况三：超时，同上

延迟队列

应用场景：下单时，提示在规定时间内完成支付，否则取消订单

实现思路：

设置超时时间并监听死信队列
给 RabbitMQ 安装延迟消息插件
- rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ
- 最多延迟两天

进入容器 docker exec -it rabbitmq /bin/bash

查看容器内插件目录对应位置 rabbitmq-plugins directories -s ，

插件目录为 /custom_plugins 或 /plugins

上传插件

启用插件

bash

# 登录进入容器内部
docker exec -it rabbitmq /bin/bash

# rabbitmq-plugins命令所在目录已经配置到$PATH环境变量中了，可以直接调用
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

# 退出Docker容器
exit

# 重启Docker容器
docker restart rabbitmq

确认插件启用

创建交换机

原本指定交换机类型的地方使用了 x-delayed-message 这个值，那么这个交换机除了支持延迟消息之外，额外使用 x-delayed-type 来指定交换机本身的类型（direct、fanout、topic）

生产者

java

@Test
public void testSendDelayMessage() {
    rabbitTemplate.convertAndSend(
            EXCHANGE_DELAY,
            ROUTING_KEY_DELAY,
            "测试基于插件的延迟消息 [" + new SimpleDateFormat("hh:mm:ss").format(new Date()) + "]",
            messageProcessor -> {

                // 设置延迟时间：以毫秒为单位
                messageProcessor.getMessageProperties().setHeader("x-delay", "10000");

                return messageProcessor;
            });
}

消费者

java

@Component  
@Slf4j
public class MyDelayMessageListener {
    
    public static final String QUEUE_DELAY = "queue.delay.video";
    
    @RabbitListener(queues = {QUEUE_DELAY})
    public void process(String dataString, Message message, Channel channel) throws IOException {  
        log.info("[生产者]" + dataString);
        log.info("[消费者]" + new SimpleDateFormat("hh:mm:ss").format(new Date()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

}

或

java

@RabbitListener(bindings = @QueueBinding(  
        value = @Queue(value = QUEUE_DELAY, durable = "true", autoDelete = "false"),  
        exchange = @Exchange(  
                value = EXCHANGE_DELAY,   
                durable = "true",   
                autoDelete = "false",   
                type = "x-delayed-message",   
                arguments = @Argument(name = "x-delayed-type", value = "direct")),  
        key = {ROUTING_KEY_DELAY}  
    ))  
    public void process(String dataString, Message message, Channel channel) throws IOException {  
        log.info("[生产者]" + dataString);  
        log.info("[消费者]" + new SimpleDateFormat("hh:mm:ss").format(new Date()));  
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);  
    }

执行效果

事务消息

事务消息和消费端没有关系，范围仅在客户端发送消息这一步，无法解决消息可靠性投递问题
在生产者端使用事务消息仅仅是控制事务内的消息是否发送
- 提交事务就把事务内所有消息都发送到交换机
- 回滚事务则事务内任何消息都不会被发送

注意要关闭生产者消息确认机制

关键配置类

java

@Configuration
public class RabbitMQConfig {

    @Bean
    public RabbitTransactionManager rabbitTransactionManager() {
        return new RabbitTransactionManager();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setChannelTransacted(true);
        return rabbitTemplate;
    }
}

生产者端

因为在 junit 使用事务，默认会回滚事务不会提交，为了避免 Junit 单元测试的影响，建议在源代码中进行实验。

java

@SpringBootApplication
public class RabbitMQProducerMainType {
    public static void main(String[] args) {
        ConfigurableApplicationContext run = SpringApplication.run(RabbitMQProducerMainType.class, args);


        MainTest mainTest = run.getBean(MainTest.class);
        mainTest.testSendMessageInTx();
    }
}

java

@Component
public class MainTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";
    public static final String ROUTING_KEY = "order";

    @Transactional
    public void testSendMessageInTx() {
        // 1、发送第一条消息
        rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, ROUTING_KEY, "I am a dragon(tx msg ~~~01)");

        // 2、抛出异常
        System.out.println("do bad:" + 10 / 0);

        // 3、发送第二条消息
        rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, ROUTING_KEY, "I am a dragon(tx msg ~~~02)");
    }
}

没有使用事务效果：RabbitMQ 收到了一条消息

使用事务效果：RabbitMQ 事务回滚，不会收到消息

惰性队列

创建队列时，在 Durability 这里有两个选项可以选择

Durable：持久化队列，消息会持久化到硬盘上
Transient：临时队列，不做持久化操作，broker 重启后消息会丢失

惰性队列概念是在持久化的前提下讨论的。惰性队列的主要特点是尽可能将消息存储在磁盘中，而不是内存中，只有在消费者消费消息时才会将消息加载到内存中。

由于各种原因，排队可能会变得很长：

消费者离线/崩溃/停机进行维护
突然出现消息进入高峰，生产者的速度超过了消费者
消费者比正常情况慢

使用惰性队列的主要原因之一是支持非常长的队列（数百万条消息），适用于消费者长时间无法消费消息的场景，使 RabbitMQ 缓存消息的能力得到提升。

优先级队列

默认情况：基于队列先进先出的特性，通常来说，先入队的先投递
设置优先级之后：优先级高的消息更大几率先投递
关键参数：x-max-priority 表示最大优先级，设置优先级超出依然为最大优先级。

创建交换机 exchange.test.priority

创建队列 queue.test.priority，指定参数 x-max-priority=10

创建路由键 routing.key.test.priority

生产者发送 1、2、3 优先级的消息

java

@Test
    public void testSendMessagePriority() {
        rabbitTemplate.convertAndSend(EXCHANGE_PRIORITY, ROUTING_KEY_PRIORITY, "I am a message with priority 3.", message->{
            message.getMessageProperties().setPriority(3);
            return message;
        });
    }

测试效果，若没有设置优先级，消息应该是先发送的先接收

集群

集群搭建

克隆三台允许 RabbitMQ 的虚拟机或运行三个 RabbitMQ Docker 容器

下面以多个 Docker 容器为例，演示 RabbitMQ 实例如何加入集群，集群节点如何彼此发现

确保容器在一个网络内（相互能 ping 通）

查看当前 RabbitMQ 节点的 Cookie 值并记录

bash

cat /var/lib/rabbitmq/.erlang.cookie
# MKHKOQNWLXOHXPZRCXWS

修改 rabbitmq-2 和 rabbitmq-3 RabbitMQ 节点的 Cookie 值为和 node01 一样：

重置节点应用并加入集群

bash

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@rabbitmq
rabbitmqctl start_app

查看集群状态

bash

rabbitmqctl cluster_status

踢出某个节点（可选）

bash

# 被踢出的节点：
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app

# 节点1
rabbitmqctl forget_cluster_node rabbit@node02

效果，三个节点都能发现彼此

管理界面负载均衡

可以配置管理界面的负载均衡实现通过统一入口访问，但实际上访问任何一个 RabbitMQ 实例的管理界面都是对集群操作。

安装 HAProxy - The Reliable, High Perf. TCP/HTTP Load Balancer

bash

apt install install -y haproxy
haproxy -v
systemctl start haproxy
systemctl enable haproxy

修改配置

bash

vim /etc/haproxy/haproxy.cfg

cfg

frontend rabbitmq_ui_frontend
bind 192.168.200.100:22222
mode http
default_backend rabbitmq_ui_backend

backend rabbitmq_ui_backend
mode http
balance roundrobin
option httpchk GET /
server rabbitmq_ui1 192.168.200.100:15672 check
server rabbitmq_ui2 192.168.200.150:15672 check
server rabbitmq_ui3 192.168.200.200:15672 check

设置 SELinux 策略，允许 HAProxy 拥有权限连接任意端口：

bash

setsebool -P haproxy_connect_any=1

重启 HAProxy

bash

systemctl restart haproxy

效果

消息功能负载均衡

haproxy 增加配置

cfg

frontend rabbitmq_frontend
bind 192.168.200.100:11111
mode tcp
default_backend rabbitmq_backend

backend rabbitmq_backend
mode tcp
balance roundrobin
server rabbitmq1 192.168.200.100:5672 check
server rabbitmq2 192.168.200.150:5672 check
server rabbitmq3 192.168.200.200:5672 check

重启 HAProxy 服务

bash

systemctl restart haproxy

创建 Rabbit 组件

交换机：exchange.cluster.test
队列：queue.cluster.test
路由键：routing.key.cluster.test

修改 RabbitMQ 服务端口为集群的负载均衡端口

yaml

spring:
  rabbitmq:
    host: 192.168.1.117
    #port: 35672
    port: 35675 # 集群的负载均衡端口

生产者

java

	private static final String EXCHANGE_CLUSTER = "exchange.cluster.test";
    private static final String ROUTING_KEY_CLUSTER = "routing.key.cluster.test";
    @Test
    public void testSendMessageCluster() {
        for (int i = 0; i < 3; i++) {
            rabbitTemplate.convertAndSend(EXCHANGE_CLUSTER, ROUTING_KEY_CLUSTER, "I am a message from producer " + i + ".");
        }
    }

消费者

java

	private static final String QUEUE_CLUSTER = "queue.cluster.test";
    @RabbitListener(queues = {QUEUE_CLUSTER})
    public void processMessageCluster(String dataString,
                                      Message message,
                                      Channel channel) throws IOException {
        log.info("消费端接收到消息内容：" + dataString);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

效果：成功收到消息

总结 Docker Compose 实践

每次构建或者重启，每个 RabbitMQ 节点和负载均衡都能顺利启动

Docker Compose 实践

yaml

services:
  rabbitmq-1:
    image: rabbitmq:4-management
    container_name: rabbitmq-1
    hostname: rabbitmq-1
    ports:
      - "35672:5672"      # 节点1 AMQP
      - "45672:15672"     # 节点1 管理界面
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: 123456
      # 插件目录（多个路径冒号分隔）
      RABBITMQ_PLUGINS_DIR: "/custom_plugins:/plugins"
      # 重要：三个节点必须使用完全相同的 Erlang Cookie
      RABBITMQ_ERLANG_COOKIE: "my-secret-cookie-value"
    volumes:
      #- ./rabbitmq-1/data:/var/lib/rabbitmq   # 每个节点独立数据目录
      - ./rabbitmq-1/custom_plugins:/custom_plugins
    networks:
      - rabbitmq-net
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
      interval: 10s      # 每10秒检查一次
      timeout: 5s        # 超时5秒
      retries: 10        # 允许10次失败（共100秒启动时间）
      start_period: 20s  # 容器启动后等待20秒再开始检查（重要！）

  rabbitmq-2:
    image: rabbitmq:4-management
    container_name: rabbitmq-2
    hostname: rabbitmq-2
    ports:
      - "35673:5672"
      - "45673:15672"
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: 123456
      RABBITMQ_PLUGINS_DIR: "/custom_plugins:/plugins"
      RABBITMQ_ERLANG_COOKIE: "my-secret-cookie-value"
    volumes:
      #- ./rabbitmq-2/data:/var/lib/rabbitmq
      - ./rabbitmq-2/custom_plugins:/custom_plugins
    networks:
      - rabbitmq-net
    restart: unless-stopped
    depends_on:
      rabbitmq-1:
        condition: service_healthy
    # 自动加入集群（等待 rabbitmq-1 完全启动后执行）
    command: >
      sh -c "
        docker-entrypoint.sh rabbitmq-server &
        sleep 20 &&
        rabbitmqctl stop_app &&
        rabbitmqctl reset &&
        rabbitmqctl join_cluster rabbit@rabbitmq-1 &&
        rabbitmqctl start_app &&
        tail -f /dev/null
      "

  rabbitmq-3:
    image: rabbitmq:4-management
    container_name: rabbitmq-3
    hostname: rabbitmq-3
    ports:
      - "35674:5672"
      - "45674:15672"
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: 123456
      RABBITMQ_PLUGINS_DIR: "/custom_plugins:/plugins"
      RABBITMQ_ERLANG_COOKIE: "my-secret-cookie-value"
    volumes:
      #- ./rabbitmq-3/data:/var/lib/rabbitmq
      - ./rabbitmq-3/custom_plugins:/custom_plugins
    networks:
      - rabbitmq-net
    restart: unless-stopped
    depends_on:
      rabbitmq-1:
        condition: service_healthy
    command: >
      sh -c "
        docker-entrypoint.sh rabbitmq-server &
        sleep 20 &&
        rabbitmqctl stop_app &&
        rabbitmqctl reset &&
        rabbitmqctl join_cluster rabbit@rabbitmq-1 &&
        rabbitmqctl start_app &&
        tail -f /dev/null
      "
  haproxy:
    image: haproxy:2.9-alpine
    container_name: haproxy
    hostname: haproxy
    ports:
      - "35675:5672"      # AMQP 负载均衡入口
      - "45675:15672"    # 管理界面负载均衡入口
    volumes:
      - ./haproxy/haproxy.cfg:/usr/local/etc/haproxy/haproxy.cfg
    networks:
      - rabbitmq-net
    depends_on:
      - rabbitmq-1
      - rabbitmq-2
      - rabbitmq-3
    restart: unless-stopped
    
networks:
  rabbitmq-net:
    driver: bridge

需提前准备 haproxy.cfg

cfg

global
    log stdout format raw local0 info
    maxconn 4096
    daemon

defaults
    log     global
    mode    tcp               # 因为 AMQP 和 HTTP 都是 TCP 协议，统一用 tcp 模式
    option  tcplog
    timeout connect 5000ms
    timeout client  50000ms
    timeout server  50000ms

# AMQP (5672) 负载均衡
frontend rabbitmq_amqp
    bind *:5672
    mode tcp
    default_backend rabbitmq_amqp_backend

backend rabbitmq_amqp_backend
    mode tcp
    balance roundrobin
    option tcp-check
    server rabbitmq-1 rabbitmq-1:5672 check inter 3s fall 3 rise 2
    server rabbitmq-2 rabbitmq-2:5672 check inter 3s fall 3 rise 2
    server rabbitmq-3 rabbitmq-3:5672 check inter 3s fall 3 rise 2

# 管理界面 (15672) 负载均衡
frontend rabbitmq_mgmt
    bind *:15672
    mode tcp
    default_backend rabbitmq_mgmt_backend

backend rabbitmq_mgmt_backend
    mode tcp
    balance roundrobin
    option tcp-check
    server rabbitmq-1 rabbitmq-1:15672 check inter 3s fall 3 rise 2
    server rabbitmq-2 rabbitmq-2:15672 check inter 3s fall 3 rise 2
    server rabbitmq-3 rabbitmq-3:15672 check inter 3s fall 3 rise 2

仲裁队列

创建仲裁队列会自动部署到所有节点

创建组件

交换机名称：exchange.quorum.test
队列名称：queue.quorum.test
路由键：routing.key.quorum.test

使用上和经典队列一样，生产消息、消费消息。

当停止某个节点的 rabbitmq 应用，集群会自动选取新的主节点，收发消息仍然正常。

Quorum queue 未来有可能取代 Classic Queue

流式队列

流式队列（Stream Queue）是 RabbitMQ 自 3.9.0 版本引入的一种新型队列类型，专为高吞吐量和大规模数据流场景设计。它通过将消息以追加日志的形式存储在磁盘上，并支持多消费者的高效读取和消息回放功能。

从客户端支持角度来说，生态尚不健全
从使用习惯角度来说，和原有队列用法不完全兼容
从竞品角度来说，像 Kafka 但远远比不上 Kafka
从应用场景角度来说：
- 经典队列：适用于系统内部异步通信场景
- 流式队列：适用于系统间跨平台、大流量、实时计算场景（Kafka 主场)
使用建议：Stream 队列在目前企业实际应用非常少，真有特定场景需要使用肯定会倾向于使用 Kafka，而不是 RabbitMQ Stream
未来展望：Classic Queue 已经有和 Quorum Queue 合二为一的趋势，Stream 也有加入进来整合成一种队列的趋势，但 Stream 内部机制决定这很难

启用 Stream 插件

bash

rabbitmq-plugins enable rabbitmq_stream

...

Federation 联邦

Federation 插件的设计目标是使 RabbitMQ 在不同的 Broker 节点之间进行消息传递而 无须建立集群。

它可以在不同的管理域中的 Broker 或 集群间传递消息，这些管理域可能设置了不同的用户和 vhost，也可能运行在不同版本的 RabbitMQ 和 Erlang 上。Federation 基于 AMQP 0-9-1 协议在不同的 Broker 之间进行通信，并且设计成能够容忍不稳定的网络连接情况。

测试环境：两个 RabbitMQ 实例（独立的broker实例），这两个 Docker 容器中的 RabbitMQ 实例不加入集群。

在上游、下游节点中都需要开启插件

bash

rabbitmq-plugins enable rabbitmq_federation
rabbitmq-plugins enable rabbitmq_federation_management

...

RabbitMQ 概述 ​

RabbitMQ 简介 ​

RabbitMQ 体系结构 ​

RabbitMQ 安装 ​

RabbitMQ 入门 ​

RabbitMQ 经典用法 ​

Work Queues 工作队列 ​

Pub/Sub 发布/订阅模式 ​

Routing 路由模式 ​

Topic 主题模式 ​

RPC 模式 ​

RabbitMQ 整合 SpringBoot ​

引入依赖 ​

消费者端 ​

@RabbitListener ​

生产者端 ​

RabbitMQ 消息可靠性投递 ​

消息发送的可靠性问题 ​

生产者消息确认机制 ​

备份交换机 ​

消息持久化 ​

消费端消息确认机制 ​

削峰限流 ​

设定 prefetch ​

消息超时 ​

死信和死信队列 ​

延迟队列 ​

事务消息 ​

惰性队列 ​

优先级队列 ​

集群 ​

集群搭建 ​

管理界面负载均衡 ​

消息功能负载均衡 ​

总结 Docker Compose 实践 ​

仲裁队列 ​

流式队列 ​

Federation 联邦 ​

RabbitMQ 概述

RabbitMQ 简介

RabbitMQ 体系结构

RabbitMQ 安装

RabbitMQ 入门

RabbitMQ 经典用法

Work Queues 工作队列

Pub/Sub 发布/订阅模式

Routing 路由模式

Topic 主题模式

RPC 模式

RabbitMQ 整合 SpringBoot

引入依赖

消费者端

@RabbitListener

生产者端

RabbitMQ 消息可靠性投递

消息发送的可靠性问题

生产者消息确认机制

备份交换机

消息持久化

消费端消息确认机制

削峰限流

设定 prefetch

消息超时

死信和死信队列

延迟队列

事务消息

惰性队列

优先级队列

集群

集群搭建

管理界面负载均衡

消息功能负载均衡

总结 Docker Compose 实践

仲裁队列

流式队列

Federation 联邦