0
0

RabbitMQ：从基础到实践

第1章 消息中间件概述

1.1 消息队列简介

消息队列(message queue)简称MQ，是一种以“先进先出”的数据结构为基础的消息服务器。

消息：在两个系统间传输的数据

作用：实现消息的传递

  • 原始的数据传递方式，如下图所示：

上述的数据传输方式为同步传输【作为调用方必须等待被调用方执行完毕以后，才可以继续传递消息】，同步传输存在的弊端：传输效率较低。

  • 基于MQ实现消息的传输，如下图所示：

上述的数据的传输方式属于异步传输【作为调用方法不用等待被调用方执行完毕就可以接续传递消息】，数据传输的消息较高。

1.2 消息队列应用场景

首先我们说一下消息中间件的主要的作用：

  • 系统解耦
  • 流量消锋
  • 数据分发

上面的三点是我们使用消息中间件最主要的目的。

1.2.1 系统解耦

系统的耦合性越高，容错性就越低。

  • 如下下图所示：以电商应用为例，用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障或者因为升级等原因暂时不可用，都会造成下单操作异常，影响用户使用体验。

  • 使用消息队列以后，整个下单操作的架构如下图所示：比如物流系统发生故障，需要几分钟才能来修复，在这段时间内，物流系统要处理的数据被缓存到消息队列中，用户的下单操作正常完成。当物流系统回复后，补充处理存在消息队列中的订单消息即可，终端系统感知不到物流系统发生过几分钟故障。

使用消息队列解耦合，系统的耦合性就会降低了，容错性就提高了。

1.2.2 流量消锋

消除系统中的高峰值流量，流量可以理解为就是请求。

  • 现有一个电商系统下单初始架构如下所示：假设用户每秒需要发送5k个请求，而我们的A系统每秒只能处理2K个请求，这样就会导致大量的下单请求失败。而且由于实际请求的数量远远超过系统的处理能力，此时也有可能导致系统宕机。

  • 使用消息队列改进以后的架构如下所示：用户每秒发送5k个请求，我们可以先将下单请求数据存储到MQ中，此时MQ中就缓存了很多下单请求数据，而A系统根据自己的处理能力从MQ中获取数据，有了MQ的缓存层以后，就可以保证每一个用户的下单请求可以得到正常的处理，并且这样可以大大提高系统的稳定性和用户体验。

1.2.3 数据分发

  • 假设A系统进行了一个业务操作，需要将操作结果通知给B、C、D系统，那么B系统、C系统、D系统就需要提供对应的接口，让A系统进行调用。
  • 如果此时不需要通知D系统了，那么就需要更改A系统的代码，将调用D系统的代码删除掉。
  • 并且如此时项目中添加了一个新的系统E，A系统也需要将处理结构通知给E系统，那么同时也需要更改A系统的代码。这样就不利于后期的维护。

原始的架构如下所示：

使用MQ改进以后的架构如下所示：

  • A系统需要将业务操作结果通知给其他系统时，A系统只需要将结果发送到MQ中。其他系统只需要从MQ中监听结果即可。
  • 如果其他系统不需要结果了，此时只需要取消从MQ中监听结果的操作即可，A系统的代码不需要进行改动。
  • 如果新增了一个系统需要获取结果，新系统只需要从MQ中监听结果就可以了，A系统的代码不需要进行改动。
  • 这样就大大的提高了系统的可维护性。

1.3 MQ的优缺点

优点：

1、应用解耦提高了系统的容错性

2、异步通讯提高了系统的响应速度

3、流量消锋提高了系统的并发能力

缺点：

1、系统可用性降低：一旦MQ宕机，就会对业务造成影响。

2、系统复杂度提高：MQ的加入大大增加了系统的复杂度。

MQ的选择依据是什么?

  • 调用方是否需要获取到被调用方的执行结果，如果需要获取到结果，那么就需要使用同步通讯，如果不需要就可以使用异步通讯。

1.4 RabbitMQ简介

RabbitMQ是由erlang【二郎神】语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。

RabbitMQ官方地址：http://www.rabbitmq.com/

RabbitMQ常见的消息模型：https://www.rabbitmq.com/getstarted.html

RabbitMQ提供了7种模式：简单模式，work模式 ，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式(远程调用)，生产者确认。

1.5 常见的消息队列产品

市场上常见的消息队列有如下：

1、ActiveMQ

2、RabbitMQ

3、RocketMQ

4、Kafka

常见特性比对：
**特性 *ActiveMQ RabbitMQRocketMQ*Kafka**开发语言javaerlangjavascala单机吞吐量万级万级10万级10万级topic数量topic可以达到几百/几千级别，吞吐量会有小幅下降，同等机器资源下，可以支撑大量的topic，这是RocketMQ的优势topic可以达到几百/几千级别，吞吐量会大幅下降，如果要支持大规模的topic，需要增加更多的机器资源时效性msus微妙级，RabbitMQ的优势，延迟最低msms级以内可用性高，基于主从架构高，基于主从架构非常高，分布式架构非常高，分布式架构，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用消息可靠性较低概率的数据丢失基本不丢失经过参数优化配置，可以做到0丢失经过参数优化配置，可以做到0丢失功支持功能完备并发能力强、性能好、延迟低，社区活跃度高，优先选择功能完善，扩展性佳，可靠性要求高的金融互联网领域使用多，经历了多次阿里双11考验功能较简单，大数据领域的实时计算以及日志采集被大规模使用

第2章 RabbitMQ环境搭建

2.1 部署RabbitMQ

本次使用Docker部署RabbitMQ，具体步骤如下所示：

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

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

2.2 部署测试

RabbitMQ后台管理系统访问地址：http://虚拟机ip地址:15672

用户名和密码：root/123

2.3 管理界面使用

2.3.1 用户管理

在创建容器的时候指定了一个使用Rabbitmq的用户root，也可以创建新的用户，如下所示：

2.3 管理界面使用

2.3.1 用户管理

在创建容器的时候指定了一个使用Rabbitmq的用户guest，也可以创建新的用户，如下所示：

常见的角色说明：

1、 超级管理员(administrator)：可登录管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。

2、 监控者(monitoring)：可登录管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)

3、 策略制定者(policymaker)：可登录管理控制台， 同时可以对policy进行管理。但无法查看节点的相关信息。

4、 普通管理者(management)：仅可登录管理控制台，无法看到节点信息，也无法对策略进行管理。

5、 其他：无法登录管理控制台，通常就是普通的生产者和消费者。

2.3.2 虚拟主机管理

在Rabbitmq中提供了一个默认的虚拟主机"/" , 也可以创建新的虚拟主机，如下所示：

2.3.3 设置访问权限

虚拟主机创建好了以后，就可以给用户设置访问该虚拟机的权限，如下所示：

第3章 RabbitMQ入门案例

需求：使用简单模式完成消息传递

3.1 父工程创建

具体步骤如下所示：

1、创建一个父工程：rabbitmq-parent

2、编写pom.xml文件：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.mq</groupId>
    <artifactId>rabbitmq-parent</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>pom</packaging>
    <name>rabbitmq-parent</name>
    <description>Parent project for RabbitMQ integration</description>

    <properties>
        <java.version>17</java.version> <!-- 根据需要修改 -->
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <spring-boot.version>3.0.5</spring-boot.version>
    </properties>

    <dependencies>
        <!-- Spring Boot 和 RabbitMQ 整合的起步依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

        <!-- Spring Boot 和 JUnit 整合的起步依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>${spring-boot.version}</version>
                <configuration>
                    <mainClass>com.mq.rabbitmqproducer.RabbitmqProducerApplication</mainClass>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

3.2 生产者工程

具体步骤如下所示：

1、在rabbitmq-parent工程下创建一个子工程rabbitmq-producer

2、编写pom.xml：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.mq</groupId>
    <artifactId>rabbitmq-producer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rabbitmq-producer</name>
    <description>rabbitmq-producer</description>
    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.6.13</spring-boot.version>
    </properties>
    <parent>
        <groupId>com.mq</groupId>
        <artifactId>rabbitmq-parent</artifactId>
        <version>0.0.1-SNAPSHOT</version>
    </parent>

    <dependencies>
        <!-- 其他依赖可以在此处添加 -->
        <!-- 子项目可以直接使用父项目中定义的依赖 -->
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <mainClass>com.mq.rabbitmqproducer.RabbitmqProducerApplication</mainClass>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

3、创建对应的启动类

package com.mq.rabbitmq_producer;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RabbitmqProducerApplication {

    public static void main(String[] args) {
        SpringApplication.run(RabbitmqProducerApplication.class, args);
    }

}

4、在application.yml文件中加入如下配置信息

spring:
  rabbitmq:
    host: 192.168.200.128
    port: 5672
    username: root
    password: 123
    virtual-host: /mq

5、编写测试类使用RabbitTemplate发送消息

@SpringBootTest
public class RabbitMQTest {

    @Autowired
    private RabbitTemplate rabbitTemplate ;

    @Test
    public void testSendMessageSimple() {
        rabbitTemplate.convertAndSend(
                "mq.queue.simple" , // 指定路由键名称
                "hello rabbitmq..." // 消息内容，也就是消息数据本身
        );
    }
}

6、创建队列

在管理后台创建一个队列 mq.queue.simple

7、启动测试，在管理后台可以发现消息存储在了队列中

3.3 消费者工程

具体步骤如下所示：

1、在rabbitmq-parent工程下创建一个子工程rabbitmq-consumer

2.直接cv rabbitmq-producer的pom文件即可

3、创建对应的启动类

package com.mq.rabbitmq;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RabbitmqConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(RabbitmqConsumerApplication.class, args);
    }

}

4、配置信息直接cv rabbitmq-parent的配置文件即可

5、编写消费者监听方法，监听指定的队列获取队列中的消息

@Component
public class MyMessageListener {

    @RabbitListener(queues = "mq.queue.simple")
    public void consumerSimple(Message message) {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("consumerSimple ...msg ----> " + msg);
    }
}

6、执行结果

7、查看管理后台可以发现消息已经被消费

3.4 架构介绍

RabbitMQ的架构图如下所示：

Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker

Virtual host:出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost

Connection：publisher／consumer 和 broker 之间的 TCP 连接

Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销

Exchange：message 到达 broker 的第一站，根据分发规则，匹配binding中的 routing key，分发消息到queue 中去。

Queue：存储消息的容器，消息最终被送到这里，等待 consumer 取走

Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。binding 信息被保存到 exchange 中，用于 message 的分发依据

第4章 RabbitMQ消息模型

4.1 简单队列模型

简单队列模型如下图所示：一个生产者对应一个消费者，这一个消费者从这个队列中消费所有的消息。

4.2 工作队列模型

4.2.1 基本介绍

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

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

4.2.2 代码演示

创建队列

在管理后台创建一个队列mq.queue.work

生产者代码
    @Test
    public void testSendMessageWork() {
        for(int x = 0 ;  x < 10 ; x++) {
            rabbitTemplate.convertAndSend(
                    "mq.queue.work" ,
                    "hello rabbitmq..." + x
            );
        }
    }
消费者代码

编写两个消费者监听器，如下所示：

 @RabbitListener(queues = "mq.queue.work")
    public void consumerWork01(Message message) {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println(
                Thread.currentThread().getName()
                        + "consumerWork01 ...msg ----> " + msg
        );
    }

    @RabbitListener(queues = "mq.queue.work")
    public void consumerWork02(Message message) {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println(
                Thread.currentThread().getName()
                        + "consumerWork02....msg ----> " + msg
        );
    }

执行结果：

org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-1consumerWork02....msg ----> hello rabbitmq...0
org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#2-1consumerWork01 ...msg ----> hello rabbitmq...1
org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#2-1consumerWork01 ...msg ----> hello rabbitmq...3
org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-1consumerWork02....msg ----> hello rabbitmq...2
org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#2-1consumerWork01 ...msg ----> hello rabbitmq...5
org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-1consumerWork02....msg ----> hello rabbitmq...4
org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-1consumerWork02....msg ----> hello rabbitmq...6
org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#2-1consumerWork01 ...msg ----> hello rabbitmq...7
org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-1consumerWork02....msg ----> hello rabbitmq...8
org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#2-1consumerWork01 ...msg ----> hello rabbitmq...9

4.3 发布订阅模型

4.3.1 基本介绍

在简单队列模型、工作队列模型中只有三个角色：

1、生产者

2、消费者

3、消息队列

不需要开发者显式的指定交换机，使用的是rabbitmq中"默认的交换机" 。

发布订阅模型中多个一个角色(exchange)，如下所示：

消息的传输过程如下所示：

1、生产者发送消息给交换机

2、交换机获取到消息将消息转发给指定的队列

3、消费者监听指定的队列，一旦队列中存在消息，消费者监听方法执行

交换机接收到消息以后到底需要将消息转发给哪一个队列，取决于交换机的类型，常见的交换机的类型：

1、Fanout：广播，将消息交给所有与之绑定队列

2、Direct：路由，把消息交给符合指定routing key 的队列

3、Topic：主题，把消息交给符合routing pattern（路由规则） 的队列

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

4.3.2 Fanout 广播

简介：fanout类型的交换机会将将消息交给所有与之绑定队列

需求：按照下图所示实现消息的传输

环境准备

要按照上图实现消息的传输，就需要先将交换机、队列、交换机和队列的绑定信息创建好。
组件组件名称交换机mq.exchange.fanout队列mq.queue.fanout01 mq.queue.fanout02

方案一： 通过后台管理系统完成

如下图所示：分别创建Exchange和Queue，然后在Exchange中绑定Queue

  • 创建交换机：要求交换机是Fanout类型

  • 创建队列

  • 创建绑定关系

方案二：通过Java Api完成

代码如下所示：在使用测试用例发送消息的时候会完相关内容的创建。

@Configuration
public class RabbitmqFanoutExchangeConfiguration {

    // 声明交换机
    @Bean
    public Exchange fanoutExchange() {
        Exchange fanoutExchange = ExchangeBuilder
                .fanoutExchange("mq.exchange.fanout")
                .durable(true).build();
        return fanoutExchange ;
    }

    // 声明队列
    @Bean
    public Queue fanoutQueue01() {
        Queue queue = QueueBuilder
                .durable("mq.queue.fanout01")
                .build();
        return queue ;
    }

    // 声明队列
    @Bean
    public Queue fanoutQueue02() {
        Queue queue = QueueBuilder
                .durable("mq.queue.fanout02")
                .build();
        return queue ;
    }

    // 声明队列和交换机的绑定信息
    @Bean
    public Binding fanoutQueue01Binding() {
        Binding binding = BindingBuilder
                .bind(fanoutQueue01())
                .to(fanoutExchange())
                .with("").noargs();
        return binding ;
    }

    @Bean
    public Binding fanoutQueue02Binding() {
        Binding binding = BindingBuilder
                .bind(fanoutQueue02())
                .to(fanoutExchange())
                .with("").noargs();
        return binding ;
    }
}
生产者代码

生产者发送消息给mq.exchange.fanout，代码如下所示：

@Test
    public void testExchangeFanout() {
        rabbitTemplate.convertAndSend(
                "mq.exchange.fanout" ,
                "" ,
                "hello fanout exchange...."
        );
    }
消费者代码

编写两个消费者，分别监听mq.queue.fanout01、mq.queue.fanout02队列，代码如下所示：

@RabbitListener(queues = "mq.queue.fanout01")
    public void consumerFanout01(Message message) {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println(
                Thread.currentThread().getName()
                        + "consumerFanout01 ...msg ----> " + msg
        );
    }

    @RabbitListener(queues = "mq.queue.fanout02")
    public void consumerFanout02(Message message) {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println(
                Thread.currentThread().getName()
                        + "consumerFanout02....msg ----> " + msg
        );
    }

执行结果：

4.3.3 Direct 定向

路由模式特点：

1、队列与交换机绑定的时候需要指定一个或者多个bindingKey(routingKey)

2、生产者发送消息的时候需要指定一个消息的routingKey

3、交换机获取到消息以后需要使用消息的routingKey和bindingKey比对，如果相等就会把消息转发给对应的队列

环境准备

组件组件名称交换机mq.exchange.direct路由键mq.routing.key.good队列mq.queue.direct

@Configuration
public class RabbitmqDirectExchangeConfiguration {

    // 声明交换机
    @Bean
    public Exchange directExchange() {
        Exchange directExchange = ExchangeBuilder.directExchange("mq.exchange.direct").durable(true).build();
        return directExchange ;
    }

    // 声明队列
    @Bean
    public Queue directQueue() {
        Queue queue = QueueBuilder.durable("mq.queue.direct").build();
        return queue ;
    }

    // 声明队列和交换机的绑定信息
    @Bean
    public Binding directQueueBinding() {
        Binding binding = BindingBuilder.bind(directQueue()).to(directExchange())
                .with("mq.routing.key.good").noargs();
        return binding ;
    }
}
生产者代码

生产者发送消息给direct_exchange，代码如下所示：

@Test
    public void testExchangeDirect() {
        rabbitTemplate.convertAndSend(
                "mq.exchange.direct" ,
                "mq.routing.key.good" ,
                "hello direct exchange....");
    }
消费者代码

编写消费者，监听mq.queue.direct，代码如下所示：

@RabbitListener(queues = "mq.queue.direct")
    public void consumerDirect(Message message) {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("consumerDirect ...msg ----> " + msg);
    }

执行结果：

4.3.4 Topic 话题

主题模式特点：

1、队列与交换机的绑定的时候需要指定一个或者多个bindingKey(routingKey) ， 在bindingKey可以使用通配符

2、生产者发送消息的时候需要指定一个消息的routingKey

3、交换机获取到消息以后需要使用消息的routingKey和bindingKey规则进行比对，如果routingKey满足bindingKey的规则就会把消息转发给对应的队列

通配符介绍：

**#**：匹配零个或多个词

**：匹配不多不少恰好1*个词

环境准备

组件组件名称交换机mq.exchange.topic路由键#.error order.* *.*队列mq.queue.message mq.queue.order
绑定关系：


    // 声明交换机
    @Bean
    public Exchange topicExchange() {
        Exchange topicExchange = ExchangeBuilder.topicExchange("mq.exchange.topic").durable(true).build();
        return topicExchange ;
    }

    // 声明队列
    @Bean
    public Queue topicQueue01() {
        Queue queue = QueueBuilder.durable("mq.queue.order").build();
        return queue ;
    }

    // 声明队列
    @Bean
    public Queue topicQueue02() {
        Queue queue = QueueBuilder.durable("mq.queue.message").build();
        return queue ;
    }

    // 声明队列和交换机的绑定信息
    @Bean
    public Binding topicQueue01Binding() {
        Binding binding = BindingBuilder.bind(topicQueue01()).to(topicExchange())
                .with("order.*").noargs();
        return binding ;
    }

    @Bean
    public Binding topicQueue02BindingErrorRoutingKey() {
        Binding binding = BindingBuilder.bind(topicQueue02()).to(topicExchange())
                .with("#.error").noargs() ;
        return binding ;
    }

    @Bean
    public Binding topicQueue02BindingInfoRoutingKey() {
        Binding binding = BindingBuilder.bind(topicQueue02()).to(topicExchange())
                .with("*.*").noargs();
        return binding;
    }
生产者代码

生产者发送消息给mq.exchange.topic，代码如下所示：

@Test
    public void testExchangeTopic() {
        rabbitTemplate.convertAndSend(
                "mq.exchange.topic",
                "order.info",
                "message order info ..."
        );
        rabbitTemplate.convertAndSend(
                "mq.exchange.topic",
                "goods.info",
                "message goods info ..."
        );
        rabbitTemplate.convertAndSend(
                "mq.exchange.topic",
                "goods.error",
                "message goods error ..."
        );
    }
消费者代码

编写两个消费者，分别监听mq.queue.message、mq.queue.order队列，代码如下所示：

@RabbitListener(queues = "mq.queue.message")
    public void consumerTopic01(Message message) {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("consumerTopic01 ...msg ----> " + msg);
    }

    @RabbitListener(queues = "mq.queue.order")
    public void consumerTopic02(Message message) {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("consumerTopic02....msg ----> " + msg);
    }

执行结果：

第5章 @RabbitListener注解

RabbitListener注解用来声明消费者监听器，可以监听指定的队列，同时也可以声明队列、交换机、队列和交换机绑定信息。

代码如下所示：

@Component
public class ConsumerListenerAnnotation {

    // @RabbitListener 注解用于定义一个 RabbitMQ 消息监听器。
    // 这个方法会监听指定的队列，并在收到消息时执行。
    // @QueueBinding 用于绑定队列和交换机，并指定路由键。
    @RabbitListener(bindings = @QueueBinding(
            // @Queue 注解用于声明一个队列。
            // value 是队列的名称 "mq.queue.test01"。
            // durable = "true" 表示该队列是持久化的，即使 RabbitMQ 服务器重启，队列仍然存在。
            value = @Queue(
                    value = "mq.queue.test01",
                    durable = "true"
            ),
            // @Exchange 注解用于声明一个交换机。
            // value 是交换机的名称 "mq.exchange.test"。
            // durable = "true" 表示交换机是持久化的。
            // type = ExchangeTypes.DIRECT 表示这是一个直连交换机，它根据路由键精确匹配消息到队列。
            exchange = @Exchange(
                    value = "mq.exchange.test",
                    durable = "true",
                    type = ExchangeTypes.DIRECT
            ),
            // key 表示路由键，用于匹配消息到队列。这里的路由键是 "info"。
            key = {"info"}
    ))
    public void consumer01(Message message) {
        // 获取消息体（消息的内容是以字节数组的形式存储的）。
        byte[] body = message.getBody();
        // 将字节数组转换为字符串，即消息的内容。
        String msg = new String(body);
        // 打印当前线程名和接收到的消息。这样可以方便地看到是哪个线程处理了该消息。
        System.out.println(Thread.currentThread().getName() + " -- direct_queue_test_01 msg：" + msg);
    }

    // 第二个 @RabbitListener 注解，监听另一个队列 "mq.queue.test02"。
    // 它使用同一个交换机 "mq.exchange.test"，但是路由键不同，这里是 "error"。
    @RabbitListener(bindings = @QueueBinding(
            // 声明一个新的队列 "mq.queue.test02"，也设置为持久化队列。
            value = @Queue(
                    value = "mq.queue.test02",
                    durable = "true"
            ),
            // 继续使用同一个交换机 "mq.exchange.test"。
            exchange = @Exchange(
                    value = "mq.exchange.test",
                    durable = "true",
                    type = ExchangeTypes.DIRECT
            ),
            // 路由键设置为 "error"，即只匹配带有路由键 "error" 的消息。
            key = {"error"}
    ))
    public void consumer02(Message message) {
        // 同样地，获取消息体，将字节数组转换为字符串。
        byte[] body = message.getBody();
        String msg = new String(body);
        // 打印当前线程名和消息内容。
        System.out.println(Thread.currentThread().getName() + " -- direct_queue_test_02 msg：" + msg);
    }
}
  • @RabbitListener: 表示当前方法是RabbitMQ的消息监听器，处理来自队列的消息。
  • bindings = @QueueBinding: 绑定队列和交换机。
  • value = @Queue(value = "mq.queue.test01", durable = "true"): 定义要绑定的队列的名称，且队列持久化。持久化指RabbitMQ服务器重启，队列也会被保留。
  • exchange = @Exchange(value = "mq.exchange.test", durable = "true", type = ExchangeTypes.DIRECT): 定义要绑定的交换机的名称，并指定交换机持久化为true，以及交换机的类型为DIRECT。交换机负责将消息路由到相应的队列。
  • key = {"info"}: 定义要绑定的路由键（routing key），路由键用于将消息发送到特定的队列。这个例子中，这个监听器会接收所有路由键为"info"的消息。

第6章 消息的可靠性保证

概述：指的就是在整个消息的传输过程中如何保证消息不丢失！

6.1 消息传输的过程回顾

Rabbitmq消息的传输过程如下所示：

上述的消息传输过程大致可以分为三个阶段：

1、生产者发送消息到MQ

2、MQ服务端存储消息

3、消费者从MQ中消费消息

在整个消息的传输过程中哪些地方可能会导致消息的丢失呢?

1、生产者发送消息到MQ：交换机的名字写错了、routingKey写错了

2、MQ服务端存储消息: MQ服务器宕机了(默认情况下消息是存储于内存中)

3、消费者从MQ中消费消息：消费者获取到消息以后还没有及时处理，消费者服务宕机了（最终消费者端的业务流程没有执行完整）

6.2 消息可靠性投递

6.2.1 可靠性方案说明

生产者发送消息的时候有两个阶段：

1、生产者发送消息到exchange

2、交换机获取到消息以后把消息转发到队列中

针对以上的两个阶段Rabbitmq提供了两种机制保障消息的可靠性投递：

confirm****确认模式：可以通过该机制确认消息是否可以正常发送到exchange

return****退回模式：可以通过该机制确认消息是否正常发送到队列中

6.2.2 可靠性代码演示

confirm机制

如下所示：

1、在配置文件中开启生产者确认机制

spring:
  rabbitmq:
    publisher-confirm-type: correlated      # 开启生产者确认机制

2、发送消息

    @Test
    public void testPublisherConfirm() {
        //通过CorrelationData对象进行封装，用于跟踪消息的状态或进行消息的确认。
        //模拟数据库生成的id
        String msgId = UUID.randomUUID().toString().replace("-", "");
        CorrelationData correlationData = new CorrelationData(msgId) ;

        // 发送消息（指定一个错误的交换机）
        rabbitTemplate.convertAndSend(
                "mq.exchange.test2" ,
                "error" ,
                "hello direct exchange...." ,
                correlationData);
    }

3、自定义RabbitTemplate，并为其绑定确认机制的回调函数，让生产者可以感知到消息是否正常投递给了交换机

@Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {

        //创建一个rabbitTemplate对象
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        //绑定生产者确认机制的回调函数
        //correlationData：封装了消息数据，包含唯一id
        //success：交换机是否接收到消成功
        //cause：交换机拒绝接收消息的原因
        rabbitTemplate.setConfirmCallback((correlationData, success, cause) -> {

            //获取当前消息的id：
            String msgId = correlationData.getId();
            if (success) {
                System.out.println("消息成功发送给交换机，msgId = " + msgId);
            } else {
                System.out.println("消息发送给交换机失败，msgId = " + msgId);
                //TODO 消息重发
            }
        });
        return rabbitTemplate;
    }

执行结果：

return机制

如下所示：

1、在配置文件中开启生产者回退机制

spring:
  rabbitmq:
    publisher-returns: true                 # 开启生产者回退机制

2、自定义RabbitTemplate，并为其绑定回退机制的回调函数，让生产者可以感知到消息是否正常投递给了队列

@Configuration
public class RabbitTemplateConfiguration {

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {

        //创建一个rabbitTemplate对象
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        //绑定生产者确认机制的回调函数
        /*rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {

            }
        });*/

        //correlationData：封装了消息数据，包含唯一id
        //ack：交换机是否接收到消息的确认字符
        //cause：交换机拒绝接收消息的原因
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {

            //获取当前消息的id：
            String msgId = correlationData.getId();
            if (ack) {
                System.out.println("消息成功发送给交换机，msgId = " + msgId);
            } else {
                System.out.println("消息发送给交换机失败，msgId = " + msgId);
            }
        });

        // 设置返回机制，处理消息无法路由到队列的情况
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnsCallback(returnedMessage -> {
            System.out.println("消息投递给队列失败，msg ---> " + returnedMessage.getMessage());
            System.out.println("replyCode ---> " + returnedMessage.getReplyCode());
            System.out.println("replyText ---> " + returnedMessage.getReplyText());
            System.out.println("exchange ---> " + returnedMessage.getExchange());
            System.out.println("routingKey ---> " + returnedMessage.getRoutingKey());
        });
        return rabbitTemplate;
    }
}

执行结果：

6.3 消息可靠性存储

针对MQ服务端存储消息导致消息丢失的情况，那么此时只需要对如下的对象进行持久化即可。

持久化：存储在消息队列中未被消费的消息，服务器重启后依然存在。

1、消息开启持久化

2、队列开启持久化

3、交换机开启持久化

  • 交换机持久化：

非持久化的交换机重启后不会被保留

  • 队列持久化：durable:true

非持久化的队列重启后不会被保留

  • 消息持久化：delivery_mode:2

非持久化的消息重启后不会被保留

设置消息为非持久化消息：

    @Test
    public void testMessageTransient() {
        rabbitTemplate.convertAndSend(
                "mq.exchange.direct",
                "mq.routing.key.good",
                "hello direct exchange....",
                (message) -> {
                    MessageProperties messageProperties = message.getMessageProperties();
                    // 设置该消息为非持久化消息
                    messageProperties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
                    return message;
                });
    }

6.4 消息可靠性消费

6.4.1 ack

在 RabbitMQ 中，

ack

（acknowledgement）和 

nack

（negative acknowledgement）是消息确认机制的一部分，用于确保消息的可靠传递。这些机制帮助 RabbitMQ 确保消息不会丢失并可以在发生错误时进行处理。

ack（确认）
  • 定义:- ack 是 “acknowledgement”（确认）的缩写。它表示消费者已经成功处理了消息。
  • 功能:- 当消费者完成对消息的处理后，它会向 RabbitMQ 发送一个 ack 确认消息已经被处理。这告诉 RabbitMQ 该消息可以从队列中移除。确认操作可以确保消息在被处理后不会重复消费。
nack（拒绝）
  • 定义:- nack 是 “negative acknowledgement”（负确认）的缩写。它表示消费者未能成功处理消息，并且该消息需要被重新处理或丢弃。
  • 功能:- 当消费者无法处理消息时，可以向 RabbitMQ 发送 nack。这通知 RabbitMQ 该消息处理失败，需要重新处理或将其移到死信队列（如果配置了死信交换机）。适用于消息处理失败的情况。nack 可以确保消息不会被丢弃而是被重新放回队列中进行重试。
ack与 nack 的重要性
  • 可靠性:- 确保消息处理的可靠性，防止消息丢失或处理失败导致的重复处理。
  • 故障恢复:- nack 允许系统在处理失败时进行故障恢复，通过重试机制确保消息最终被成功处理。

通过适当地使用 

ack

和 

nack

，可以确保 RabbitMQ 消息系统的可靠性和健壮性，使消息的处理变得更为稳定和可控

6.4.2 消费消息流程

消费者获取到消息以后需要给RabbitMQ服务端进行应答，RabbitMQ根据消费者的应答信息决定是否需要将消息从RabbitMQ的服务端删除掉。

应答模式：

1、none： 消费者在处理消息后不需要向 RabbitMQ 发送任何确认。这意味着消息一旦被投递到消费者，无论消费者是否成功处理了消息，RabbitMQ 都直接将消息从队列中移除。

2、auto(默认值)：自动应答，如果消费正常，则删除消息，返回ack；如果消费异常，则不删除消息，返回nack。

3、manual： 手动应答，消费者可以根据消息消费的实际情况给RabbitMQ进行应答。如果消费正常，返回ack；如果消费异常，返回nack。

6.4.3 none模式

在消费者的application.yml文件中，设置消息者应答模式为none模式，如下所示：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: none
环境准备

组件组件名称交换机direct_exchange路由键error队列direct_queue_01

监听器
@RabbitListener(queues = "direct_queue_01")
public void consumerAcknowledgeMode(Message message) {
    byte[] body = message.getBody();
    String msg = new String(body);
    System.out.println("consumer01 ...msg ----> " + msg + " " + new Date());
    int a = 1 / 0 ;
}
生产者
@Test
public void testAcknowledgeMode(){

    String msgId = UUID.randomUUID().toString().replace("-", "");//模拟数据库生成的id
    CorrelationData correlationData = new CorrelationData(msgId) ;

    rabbitTemplate.convertAndSend(
        "direct_exchange",
        "error",
        "direct exchange testAcknowledgeMode...",
        correlationData);
}

执行结果：

疯狂报错，正确。

6.4.4 auto模式

在消费者的application.yml文件中，设置消息者应答模式为auto模式并再次测试，如下所示：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto    # 更改消费者应答模式为自动模式

即使是业务代码产生了异常，消息不会从RabbitMQ服务端删除掉，但是出现了无限次消费的情况。

解决方案：在application.yml文件中添加如下的配置即可

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto      # 更改消费者应答模式为自动模式
        retry:
          enabled: true             # 开启消费者(程序出现异常的情况下,捕获异常重试将不生效)进行重试
          max-attempts: 5           # 最大重试次数(包含第一次调用)
          initial-interval: 2000    # #重试间隔时间
          max-interval: 5000        # 最大间隔时间
          multiplier: 2             # 乘子  重试间隔*乘子得出下次重试间隔   2s 4s 8s>5s(取5s)

测试结果：五次之后报错就不会重试了。

注意：

1、测试重试的时候不能通过异常的输出次数来判断方法调用了几次。

2、重试次数耗尽以后会调用MessageRecoverer中的recover方法对消息进行处理。

  • RejectAndDontRequeueRecoverer（默认）：拒绝而且不把消息重新放入队列，也就是说消息会丢失
  • RepublishMessageRecoverer：重新发布消息
  • ImmediateRequeueMessageRecoverer：立即把消息重新放入队列

6.4.5 manual模式

当重试次数耗尽以后，auto模式消息的处理还是失败，直接将消息从RabbitMQ服务端删除掉，相当于消息丢失。那么针对这种情况最好使用manual模式。

在消费者的application.yml文件中，设置消息者应答模式为manual模式并再次测试，如下所示：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual      # 更改消费者应答模式为manual模式

测试1：无论有没有异常发生，如果没有手动进行ack确认，那么消息不会被消费，需要进行手动ack确认

测试2：有异常发生，则根据重试配置发起重试

手动ack代码如下：

@RabbitListener(queues = "direct_queue_01")
public void consumer01(Message message , Channel channel) throws IOException {

    byte[] body = message.getBody();
    String msg = new String(body);

    System.out.println("consumer01 ...msg ----> " + msg + " " + new Date());
    int a = 1 / 0 ;
    // 如果正常消费，则返回应答ack；如果非正常消费，则根据重试配置发起重试
    // 消息标签，标识消息的唯一性
    long deliveryTag = message.getMessageProperties().getDeliveryTag();    
    channel.basicAck(deliveryTag , true);
}

注意：业务代码如果捕获了业务中抛出的异常，重试机制将不会生效

解决方案：设置最大重试次数（手动实现）、需要配合Redis。

@RabbitListener(queues = "direct_queue_01")
public void consumer01(Message message , Channel channel){

    byte[] body = message.getBody();
    String msg = new String(body);

    // 消息标签，标识消息的唯一性
    long deliveryTag = message.getMessageProperties().getDeliveryTag();     
    try {
        System.out.println("consumer01 ...msg ----> " + msg + " " + new Date());
        int a = 1 / 0 ;
        // 如果正常消费，则返回应答ack；如果非正常消费，则根据重试配置发起重试
        // 参数1：消息标签，标识消息的唯一性
        // 参数2：true：确认所有小于等于当前deliveryTag的消息；false：仅确认当前deliveryTag对应的消息
        channel.basicAck(deliveryTag , true);

    }catch (Exception e) { //注意：此处需要是Exception，能够捕获到除零异常
        e.printStackTrace();
        try {
            // 非正常消费返回应答ack
            //第一个true：拒绝所有未确认的消息；false：只拒绝当前未确认的消息
            //第二个true：消息是否重新投递以便重新排队；false：丢弃或放入死信队列（如果已设置）
            channel.basicNack(deliveryTag , true , true); 
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}

第7章 消费者限流

默认情况下RabbitMQ会将所有的消息投递给对应的队列，不会考虑消费者的消费能力。但是在实际的开发过程中，需要根据消费者的消费能力来处理队列中的消息。

如下图所示：

具体实现：在消费者的application.yml配置文件中加入如下的配置

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual      # 更改消费者应答模式为自动模式
        prefetch: 10                  # Rabbitmq服务端一次投递10个消息给消费者，当10个消息应答完毕以后再投递10个消息过来

代码测试：

监听器：

@RabbitListener(queues = "direct_queue_01")
public void consumer01(Message message , Channel channel) throws IOException, InterruptedException {

    //处理消息慢一点
    TimeUnit.MILLISECONDS.sleep(500);

    byte[] body = message.getBody();
    String msg = new String(body);
    System.out.println("consumer01 ...msg ----> " + msg + " " + new Date());

    // 如果正常消费，则返回应答ack；如果非正常消费，则根据重试配置发起重试
    long deliveryTag = message.getMessageProperties().getDeliveryTag(); // 消息标签，标识消息的唯一性
    channel.basicAck(deliveryTag , true);
}

​​​​​​​生产者：

@Test
public void test01() {
    for (int i = 0; i < 100; i++) {
        rabbitTemplate.convertAndSend(
                "direct_exchange" ,
                "error" ,
                "msg...." + i);
    }
}

如下图，可以看到“待处理”消息和“待ack”消息相加是“总消息”，每次处理完毕等待ack返回的是10条消息，说明每10条消息一处理

第8章 消息存活时间

TTL 全称 Time To Live（存活时间/过期时间）。

当消息在队列中时间到达存活时间后，还没有被消费，会被自动清除。

RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。

给消息设置存活时间：

给队列设置消息存储时间：

讲解消息存活时间的目的是为后面讲解延迟队列做铺垫。

第9章 死信队列

9.1 死信

死信：死掉的消息

消息成为死信的三种情况

1、队列消息数量到达限制；比如队列最大只能存储10条消息，而发了11条消息，根据先进先出，最先发的消息成为死信。

2、消费者发送了basicNack/basicReject，并且不把消息重新放入原目标队列，requeue=false；

3、原队列存在消息存活时间设置，消息到达存活时间未被消费；

注意：默认情况下Rabbitmq会直接将死信丢弃掉，但是如果在系统中提供了死信队列，那么此时就会把消息投递给死信队列。

9.2 死信队列

概述：存储死信的队列

针对一些不重要的消息，成为死信以后使用默认的机制(丢弃)是可以的，但是针对比较重要的消息，此时就不能丢弃，就需要将这个死信存储起来，这时候就需要使用到死信队列。

在存储死信到死信队列的时候，需要使用到死信交换机：

DeadLetter Exchange（死信交换机），英文缩写：DLX 。当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX。后期这个交换机就可以将消息投递到与之绑定的死信队列中。

如下图所示：

9.3死信队列演示

代码如下所示：

注意：测试之前先删除direct_queue_01，然后启动生产者端发送消息，让新的元素创建出来

@Bean
    public Exchange directExchange() {
        return ExchangeBuilder.directExchange("direct_exchange").durable(true).build();
    }

    // 声明死信交换机
    @Bean
    public Exchange dlxExchange() {
        return ExchangeBuilder.directExchange("dlx_exchange").durable(true).build();
    }

    // 声明死信队列
    @Bean
    public Queue dlxQueue() {
        return QueueBuilder.durable("dlx_queue").build();
    }

    // 声明死信交换机和死信队列绑定信息
    @Bean
    public Binding dlxQueueBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("dlx").noargs();
    }

    // 声明主队列，并且绑定死信交换机
    @Bean
    public Queue directQueue() {
        return QueueBuilder.durable("direct_queue_01")
                .maxLength(10)
                .deadLetterExchange("dlx_exchange")
                .deadLetterRoutingKey("dlx")
                .build();
    }

    // 将主队列绑定到主交换机
    @Bean
    public Binding directQueueBinding() {
        return BindingBuilder.bind(directQueue()).to(directExchange()).with("error").noargs();
    }

生产者：代码同前

@Test
public void test01() {
    rabbitTemplate.convertAndSend("direct_exchange" , "error" , "msg....");
}

监听器：代码同前

@RabbitListener(queues = "direct_queue_01")
public void consumer01(Message message , Channel channel){

    byte[] body = message.getBody();
    String msg = new String(body);

    // 消息标签，标识消息的唯一性
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    try {
        System.out.println("consumer01 ...msg ----> " + msg + " " + new Date());
        int a = 1 / 0 ;
        // 如果正常消费，则返回应答ack；如果非正常消费，则根据重试配置发起重试
        // 参数1：消息标签，标识消息的唯一性
        // 参数2：true：确认所有小于等于当前deliveryTag的消息；false：仅确认当前deliveryTag对应的消息
        channel.basicAck(deliveryTag , true);

    }catch (Exception e) {
        e.printStackTrace();
        try {
            // 非正常消费返回应答ack
            //第一个true：拒绝所有未确认的消息；false：只拒绝当前未确认的消息
            //第二个true：消息是否重新投递以便重新排队；false：丢弃或放入死信队列（如果已设置）
            channel.basicNack(deliveryTag , true , false);
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}

死信队列处理消息

@RabbitListener(queues = "dlx_queue")
public void consumerDlx(Message message, Channel channel) throws IOException {

    byte[] body = message.getBody();
    String msg = new String(body);
    System.out.println("consumerDlx ...msg ----> " + msg);
    // 消息标签，标识消息的唯一性
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    channel.basicAck(deliveryTag , true);
}

第10章 延迟队列

10.1 延迟队列简介

延迟队列存储的对象肯定是对应的延时消息，所谓延时消息是指当消息被发送以后，并不想让消费者立即拿到消息，而是等待指定时间后，消费者才拿到这个消息进行消费。

场景：在订单系统中，一个用户下单之后通常有30分钟的时间进行支付，如果30分钟之内没有支付成功，那么这个订单将进行取消处理。这时就可以使用延时队列将订单信息发送到延时队列。

很可惜，在RabbitMQ中并未提供延迟队列功能。但是可以使用：ttl +死信队列 组合实现延迟队列的效果。

架构如下所示：

10.2 延迟队列演示

10.2.1 延迟队列声明

代码如下所示：

注意：测试之前先删除direct_queue_01，然后启动生产者端发送消息，让新的元素创建出来

@Bean
   public Exchange directExchange() {
       return ExchangeBuilder.directExchange("direct_exchange").durable(true).build();
   }

    // 声明死信交换机
    @Bean
    public Exchange dlxExchange() {
        return ExchangeBuilder.directExchange("dlx_exchange").durable(true).build();
    }

    // 声明死信队列
    @Bean
    public Queue dlxQueue() {
        return QueueBuilder.durable("dlx_queue").build();
    }

    // 声明死信交换机和死信队列绑定信息
    @Bean
    public Binding dlxQueueBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("dlx").noargs();
    }

    // 声明主队列，并且绑定死信交换机
    @Bean
    public Queue directQueue01() {
        return QueueBuilder.durable("direct_queue_01")
                .ttl(5000) // 消息的存活时间
                .deadLetterExchange("dlx_exchange")
                .deadLetterRoutingKey("dlx")
                .build();
    }

    // 将主队列绑定到主交换机
    @Bean
    public Binding directQueueBinding() {
        return BindingBuilder.bind(directQueue01()).to(directExchange()).with("error").noargs();
    }

10.2.2 生产者代码

生产者发送消息给正常交换机，代码同前:

@Test
public void test01() {
    rabbitTemplate.convertAndSend("direct_exchange" , "error" , "msg....");
}

10.2.3 消费者代码

消费者监听死信队列，正常队列不需要绑定消费者，代码如下所示：

注意：删除direct_queue_01的监听

@RabbitListener(queues = "dlx_queue")
public void consumer01(Message message , Channel channel) {

    // 获取消息相关信息
    long deliveryTag = message.getMessageProperties().getDeliveryTag();     // 消息标签，标识消息的唯一性
    byte[] body = message.getBody();
    String msg = new String(body);
    try {

        System.out.println("consumer01Listener ...msg ----> " + msg);
        channel.basicAck(deliveryTag , true);           // 正常消费返回应答ack

    }catch (Exception e) {
        e.printStackTrace();
        try {
            channel.basicNack(deliveryTag , true , true);      // 非正常消费返回应答ack
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}

第11章 消息的重复消费问题

11.1 消息百分百成功投递架构

针对一些特殊的业务，要严格保证消息能够进行正常传输。那么此时在进行消息投递的时候，就可以使用如下的架构保证消息百分百成功投递：

Step 1： 首先把消息信息(业务数据）存储到数据库中，紧接着，我们再把这个消息记录也存储到一张消息记录表里（或者另外一个同源数据库的消息记录表），并且在消息数据库表中需要指定一个状态字段status来记录消息的投递状态。

Step 2：发送消息到MQ Broker节点（采用confirm方式发送，会有异步的返回结果）

Step 3、4：生产者端接受MQ Broker节点返回的Confirm确认消息结果，然后更新消息记录表里的消息状态。比如默认Status = 0 当收到消息确认成功后，更新为1即可！

Step 5：但是在消息确认这个过程中可能由于网络闪断、MQ Broker端异常等原因导致 回送消息失败或者异常。这个时候就需要发送方（生产者）对消息进行可靠性投递了，保障消息不丢失，100%的投递成功！（有一种极限情况是闪断，Broker返回的成功确认消息，但是生产端由于网络闪断没收到，这个时候重新投递可能会造成消息重复，需要消费端去做幂等处理）所以我们需要有一个定时任务，（比如每5分钟拉取一下处于中间状态的消息，当然这个消息可以设置一个超时时间，比如超过1分钟 Status = 0 ，也就说明了1分钟这个时间窗口内，我们的消息没有被确认，那么会被定时任务拉取出来）

Step 6：接下来我们把中间状态的消息进行重新投递 retry send，继续发送消息到MQ ，当然也可能有多种原因导致发送失败

Step 7：我们可以采用设置最大努力尝试次数，比如投递了3次，还是失败，那么我们可以将最终状态设置为Status = 2 ，最后 交由人工解决处理此类问题（或者把消息转储到失败表中）。

情况一：投递过程中产生了网络抖动不会导致消息丢失(因为消息已经入库)

情况二：confirm回执的时候产生了网络抖动不会导致消息丢失(因为消息已经入库)

数据库表设计：

DROP TABLE IF EXISTS `broker_message_log`;
CREATE TABLE `broker_message_log` (
  `message_id` varchar(255) NOT NULL COMMENT '消息唯一ID',
  `message` varchar(4000) NOT NULL COMMENT '消息内容',
  `try_count` int(4) DEFAULT '0' COMMENT '重试次数',
  `status` varchar(10) DEFAULT '' COMMENT '消息投递状态 0投递中,1投递成功,2投递失败',
  `next_retry` timestamp NOT NULL DEFAULT '0000-00-00 00：00：00' ON UPDATE CURRENT_TIMESTAMP COMMENT '下一次重试时间',
  `create_time` timestamp NOT NULL DEFAULT '0000-00-00 00：00：00' ON UPDATE CURRENT_TIMESTAMP,
  `update_time` timestamp NOT NULL DEFAULT '0000-00-00 00：00：00' ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`message_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- ----------------------------
-- Table structure for t_order
-- ----------------------------
DROP TABLE IF EXISTS `t_order`;
CREATE TABLE `t_order` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(255) DEFAULT NULL,
  `message_id` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2018091102 DEFAULT CHARSET=utf8;

11.2 消息的重复消费问题

11.2.1 问题说明

采用上述架构实现消息的投递，那么此时就会出现消息的重复消费问题。

MQ中出现了重复消息，那么此时就会导致重复消费问题。在有一些特殊的业务场景下，是不允许出现重复消息的，比如扣减库存。

11.2.2 幂等性处理

幂等性指一次和多次请求某一个资源对于资源本身应该具有同样的结果。也就是说,其任意多次执行对资源本身所产生的影响均与一次执行的影响相同。

在MQ中指,消费多条相同的消息,得到与消费该消息一次相同的结果。

消息幂等性保障 ：数据库唯一字段

标签： rabbitmq 分布式 java
本文转载自: https://blog.csdn.net/weixin_67796933/article/details/142374466
版权归原作者 贰陆.256 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ：从基础到实践”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

雷池WAF自动化实现安全运营实操案例终极篇

以太网交换安全：MAC地址漂移

安全见闻4（泷羽sec）

爬虫-selenium_edge的无界面模式

Git代码管理规范

【微服务】认识微服务及Eureka注册中心

Nginx 实现会话保持的方式配置

文章导航