RabbitMQ从入门到精通

╰半橙微兮° 2023-10-02 21:19 2阅读 0赞

### 史上最全RabbitMQ讲解！ ###

**目录**

史上最全RabbitMQ讲解！（持续更新）

一.RabbitMQ概述

1.消息队列的优点

2.消息队列的缺点

3.MQ如何选型

二.RabbitMQ的重要概念

1.Publisher

2.Message

3.Exchange

4.BindingKey

5.Routingkey

6.Routingkey和BindingKey的区别

7.Queue

8.Consumer

9.Connection

10.Channel

11.Virtual Host

12.Broker

三.RabbitMQ发送和接收消息步骤

1.发送消息的步骤

2.消费者接收消息

四.RabbitMQ的四种交换机

1.直连交换机（Direct exchange）

2.扇形交换器（Fanout exchange）

3.主题交换器（Topic exchange）

4.头信息交换器（Headers exchange）

五.RabbitMQ的六种工作模式

1.simple简单模式

2.work工作模式 （竞争关系）

3.publish/subscribe发布订阅(共享资源)

4.routing路由模式

5.topic 主题模式(路由模式的一种)

6.远程过程调用（RPC）

总结

六.Springboot集成RabbitMQ

1.生产者搭建

2.消费者

七.RabbitMQ之延时队列

1.什么是延时队列

2.延时队列和定时任务的区别

3.RabbitMQ怎么实现延时队列

4.TTL(消息过期时间)

5.死信队列

八.RabbitMQ消息可靠性保障

1.RabbiMQ消息处理逻辑

2.保证RabbiMQ消息可靠性的方式

3.生产者保证消息可靠性

4.Broker保证消息可靠性

5.消费者保证消息可靠性

九.RabbitMQ业务可靠性分析

幂等性校验

数据回滚

--------------------

### 一.RabbitMQ概述 ###

#### 1.消息队列的优点 ####

**（1）应用解耦**

用户下单成功后，短信通知用户。这个发短信的功能可以使用mq实现。

**（2）异步**

用户下单后，立马返回订单详情。发短信是异步的。

**（3）流量削峰**

当下单用户量达到40W/s，但是我们业务处理能力只有20W/s，剩余20W可以放到mq中慢慢处理。mq起到一个缓冲的作用。

#### 2.消息队列的缺点 ####

**（1）系统可用性降低**

消息队列在系统中充当一个中间人的身份，如果该中间人突然失联了，那其他两方就不知所措了，最后也就导致系统之间无法互动。

**（2）系统复杂性提高**

在使用消息队列的过程中，难免会出现生产者、MQ、消费者宕机不可用的情况，那么随之带来的问题就是消息重复、消息乱序、消息堆积等等问题都需要我们来控制。

**（3）消息一致性问题**

如下图所示，系统需要保证快递投递，扣减系统费，通知等之间的数据一致性，如果系统短信通知，快递通知执行成功，扣减系统费执行失败时，就会出现数据不一致问题

#### 3.MQ如何选型 ####

**（1）各种MQ选型分析**

**RabbitMQ**：erlang开发，对消息堆积的支持并不好，当大量消息积压的时候，会导致 RabbitMQ 的性能急剧下降。每秒钟可以处理几万到十几万条消息。

**RocketMQ**：java开发，面向互联网集群化功能丰富，对在线业务的响应时延做了很多的优化，大多数情况下可以做到毫秒级的响应，每秒钟大概能处理几十万条消息。

**Kafka**：Scala开发，面向日志功能丰富，性能最高。当你的业务场景中，每秒钟消息数量没有那么多的时候，Kafka 的时延反而会比较高。所以，Kafka 不太适合在线业务场景。

**（2）个人建议**

中小型公司，技术一般，可以考虑用 RabbitMQ；

大型公司，基础架构研发实力较强，用 RocketMQ 是很好的选；

实时计算、日志采集：使用 kafka。

**（3）应用场景**

详见我的另一篇博客：[https://blog.csdn.net/qq\_45443475/article/details/119934030?spm=1001.2014.3001.5502][https_blog.csdn.net_qq_45443475_article_details_119934030_spm_1001.2014.3001.5502]

### 二.RabbitMQ的重要概念 ###

#### 1.Publisher ####

消息发布者 (或称为生产者) 负责生产消息并将其投递到指定的交换器上。

#### 2.Message ####

消息由消息头和消息体组成，消息头用于存储与消息相关的元数据：如目标交换器的名字 (exchange\_name) 、路由键 (RountingKey)和其他可选配置 (properties) 信息。消息体为实际需要传递的数据。

#### 3.Exchange ####

交换器负责接收来自生产者的消息，并将消息路由到一个或者多个队列中，如果路由不到，则返回给生产者或者直接丢弃，这取决于交换器的 mandatory 属性。

mandatory = true时， 如果交换器路由不到队列，则会将该消息返回给生产者。

mandatory = false时， 如果交换器路由不到队列，则会直接丢弃该消息。

#### 4.BindingKey ####

交换器与队列通过 BindingKey 建立绑定关系。

#### 5.Routingkey ####

基于交换器类型的规则相匹配时，消息被路由到对应的队列中。

#### **6.Routingkey和BindingKey的区别** ####

String routingKey = "class.sex" ;// 消息的路由键，例如一班.李四

String bindingKey = "\*.李\*"; // 队列绑定的接收规则, 列如只收李姓

#### 7.Queue ####

消息队列载体，每个消息都会被投入到一个或多个队列。

用于存储路由过来的消息，多个消费者可以订阅同一个消息队列，此时队列会将收到的消息将以轮询 (round-robin) 的方式分发给所有消费者，即每条消息只会发送给一个消费者，不会出现一条消息被多个消费者重复消费的情况。

#### 8.Consumer ####

消费者订阅感兴趣的队列，并负责消费存储在队列中的消息。为了保证消息能够从队列可靠地到达消费者，RabbitMQ 提供了消息确认机制 (messageacknowledgement)，并通过 autoAck 参数来进行控制。

**autoAck=true**  此时消息发送出去 (写入TCP套接字) 后就认为消费成功，而不管消费者是否真正消费到这些消息。当 TCP 连接或 channel 因意外而关闭，或者消费者在消费过程之中意外宕机时，对应的消息就丢失。因此这种模式可以提高吞吐量，但会存在数据丢失的风险。

**autoAck=false** 需要用户在数据处理完成后进行手动确认，只有用户手动确认完成后，RabbitMQ 才认为这条消息已经被成功处理，这可以保证数据的可靠性投递，但会降低系统的吞吐量。

#### 9.Connection ####

用于传递消息的 TCP 连接

#### 10.Channel ####

消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

#### 11.Virtual Host ####

虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。

RabbitMQ 通过虚拟主机来实现逻辑分组和资源隔离，一个虚拟主机就是一个小型的 RabbitMQ服务器，拥有独立的队列、交换器和绑定关系。用户可以按照不同业务场景建立不同的虚拟主机，虚拟主机之间是完全独立的，你无法将 vhost1 上的交换器与vhost2 上的队列进行绑定，这可以极大的保证业务之间的隔离性和数据安全，默认的虚拟主机名为 `/` 。

#### 12.Broker ####

消息队列所在的服务器实体。

### 三.RabbitMQ发送和接收消息步骤 ###

#### 1.发送消息的步骤 ####

（1）生产者声明一个exchange和一个queue

public static final String EXCHANGE_NAME = "boot_topic_exchange";
        public static final String QUEUE_NAME = "boot_queue";
         //1.声明交换机
        @Bean("bootExchange")
        public Exchange bootExchange(){
            return ExchangeBuilder.topicExchange(EXCHANGE_NAME)  //声明topic模式的交换机
                    .durable(true)  //开启持久化
                    .build();       //构建交换机
        }
        //2.声明队列
        @Bean("bootQueue")
        public Queue bootQueue(){
            return QueueBuilder.durable(QUEUE_NAME).build();
        }

（2）生产者使用routing key，绑定exchange和queue

//3.声明交换机和队列的绑定关系(无需注入，只写Bean注解即可)
        @Bean
        public Binding bindExchangeQueue(@Qualifier("bootExchange") Exchange exchange,
                                         @Qualifier("bootQueue") Queue queue){
            return BindingBuilder.bind(queue)   //队列
                    .to(exchange)               //交换机
                    .with("boot.#")  //路由key
                    .noargs();                  //是否还有参数
        }

#### 2.消费者接收消息 ####

@Component
    public class RabbitmqListener {
    
        @RabbitListener(queues = "boot_queue")
        public void ListenQueueMsg(Message msg){
            System.out.println("接收到消息:"+msg);
        }
    
    }

### 四.RabbitMQ的四种交换机 ###

#### 1.直连交换机（Direct exchange） ####

![f81344fa6d244c1892bf22acd9d84812.png][]

直连型交换机背后的路由算法很简单——消息会传送给绑定键与消息的路由键完全匹配的那个队列。 我们用直连交换机取代了只会无脑广播的扇形交换机，并且具备了选择性接收消息的能力。

这种配置下，我们可以看到有两个队列Q1、Q2绑定到了直连交换机X上。第一个队列用的是橘色（orange）绑定键，第二个有两个绑定键，其中一个绑定键是黑色（black），另一个绑定键是绿色（green）。在此设置中，发布到交换机的带有橘色（orange）路由键的消息会被路由给队列Q1。带有黑色（black）或绿色（green）路由键的消息会被路由给Q2。其他的消息则会被丢弃。

#### 2.扇形交换器（Fanout exchange） ####

![64280aed55184213b8cfd6ad309d8f49.png][]

当一个Msg发送到扇形交换机X上时，则扇形交换机X会将消息分别发送给所有绑定到X上的消息队列。扇形交换机将消息路由给绑定到自身的所有消息队列，也就是说路由键在扇形交换机里没有作用，故消息队列绑定扇形交换机时，路由键可为空。这个模式类似于广播。

#### 3.主题交换器（Topic exchange） ####

![c7e3048d27c44ae7ae71fbf3d761c2f1.png][]

（1）路由键和绑定键命名  
消息路由键—发送到主题交换机的消息所携带的路由键（routing\_key）不能随意命名——它必须是一个用点号分隔的词列表。当中的词可以是任何单词，不过一般都会指定一些跟消息有关的特征作为这些单词。列举几个有效的路由键的例子：“stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”。只要不超过255个字节，词的长度由你来定。  
绑定键（binding key）也得使用相同的格式。主题交换机背后的逻辑跟直连交换机比较相似——一条携带特定路由键（routing key）的消息会被投送给所有绑定键（binding key）与之相匹配的队列。尽管如此，仍然有两条与绑定键相关的特殊情况：  
\* (星号) 能够替代一个单词。  
\# (井号) 能够替代零个或多个单词。  
（2）示例解析，如上图：  
我们将会发送用来描述动物的多条消息。发送的消息包含带有三个单词（两个点号）的路由键（routing key）。路由键中第一个单词描述速度，第二个单词是颜色，第三个是品种： “<速度>.<颜色>.<品种>”。我们创建三个绑定：Q1通过".orange.“绑定键进行绑定，Q2使用”..rabbit" 和 “lazy.\#”。

队列绑定键解释：  
Q1针对所有的橘色orange动物。  
Q2针对每一个有关兔子rabbits和慵懒lazy的动物的消息。  
消息路由键解释：  
一个带有"quick.orange.rabbit"路由键的消息会给两个队列都进行投送。消息"lazy.orange.elephant"也会投送给这两个队列。  
另外一方面，“quick.orange.fox” 只会给第一个队列。"lazy.pink.rabbit"虽然与两个绑定键都匹配，但只会给第二个队列投送一遍。“quick.brown.fox” 没有匹配到任何绑定，因此会被丢弃掉。  
（3）异常情况  
如果我们破坏规则，发送的消息只带有一个或者四个单词，例如 “orange” 或者 "quick.orange.male.rabbit"会发生什么呢？结果是这些消息不会匹配到任何绑定，将会被丢弃。另一方面，“lazy.orange.male.rabbit”即使有四个单词，也会与最后一个绑定匹配，并 被投送到第二个队列。  
（4）注意事项  
主题交换机非常强大，并且可以表现的跟其他交换机相似。  
当一个队列使用"\#"（井号）绑定键进行绑定。它会表现的像扇形交换机一样，不理会路由键，接收所有消息。  
当绑定当中不包含任何一个 “\*” (星号) 和 “\#” (井号)特殊字符的时候，主题交换机会表现的跟直连交换机一样。

#### 4.头信息交换器（Headers exchange） ####

头交换机类似与主题交换机，但是却和主题交换机有着很大的不同。主题交换机使用路由键来进行消息的路由，而头交换机使用消息属性来进行消息的分发，通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。  
在头交换机里有一个特别的参数”x-match”，当”x-match”的值为“any”时，只需要消息头的任意一个值匹配成功即可，当”x-match”值为“all”时，要求消息头的所有值都需相等才可匹配成功。

### 五.RabbitMQ的六种工作模式 ###

#### 1.simple简单模式 ####

![59921272f5f24e0098dfd801cd49e038.png][]

#### 2.work工作模式 （竞争关系） ####

![151c20fec79943cd94adc4be0ca6955b.png][]

消息产生者将消息放入队列消费者可以有多个,消费者1,消费者2,同时监听同一个队列,消息被消费?C1 C2共同争抢当前的消息队列内容,谁先拿到谁负责消费消息(隐患,高并发情况下,默认会产生某一个消息被多个消费者共同使用,可以设置一个开关(syncronize,与同步锁的性能不一样) 保证一条消息只能被一个消费者使用)  
应用场景:红包;大项目中的资源调度(任务分配系统不需知道哪一个任务执行系统在空闲,直接将任务扔到消息队列中,空闲的系统自动争抢)

#### 3.publish/subscribe发布订阅(共享资源) ####

![bf6a1333d867423e9d58654cc2e5b85c.png][]

X代表交换机rabbitMQ内部组件,erlang 消息产生者是代码完成,代码的执行效率不高,消息产生者将消息放入交换机,交换机发布订阅把消息发送到所有消息队列中,对应消息队列的消费者拿到消息进行消费  
相关场景:邮件群发,群聊天,广播(广告)  
发布订阅用的交换机类型为fanout

#### 4.routing路由模式 ####

![0a4dd7df68964dd588521b7ebf781455.png][]

消息生产者将消息发送给交换机按照路由判断,路由是字符串(info) 当前产生的消息携带路由字符(对象的方法),交换机根据路由的key,只能匹配上路由key对应的消息队列,对应的消费者才能消费消息;  
根据业务功能定义路由字符串  
从系统的代码逻辑中获取对应的功能字符串,将消息任务扔到对应的队列中业务场景:error 通知;EXCEPTION;错误通知的功能;传统意义的错误通知;客户通知;利用key路由,可以将程序中的错误封装成消息传入到消息队列中,开发者可以自定义消费者,实时接收错误;  
交换机类型：direct

#### 5.topic 主题模式(路由模式的一种) ####

![bf3b46a7871c439a80922c214b37c453.png][]星号井号代表  
星号代表一个单词,井号代表多个单词  
路由功能添加模糊匹配  
消息产生者产生消息,把消息交给交换机  
交换机根据key的规则模糊匹配到对应的队列,由队列的监听消费者接收消息消费  
交换机类型：topic

####    6.远程过程调用（RPC） ####

![f7c170efe1ec4ab8a70560fe47c8a21b.png][]

1、使用RabbitMQ构建RPC系统：客户端和可伸缩的RPC服务器。由于我们没有任何值得分发的耗时任务，我们将创建一个返回Fibonacci数字的虚拟RPC服务。  
2、对于RPC请求，客户端发送带有两个属性的消息： replyTo，设置为仅为请求创建的匿名独占队列;以及correlationId，设置为每个请求的唯一值。  
3、客户端等待回复队列上的数据。出现消息时，它会检查correlationId属性。如果它与请求中的值匹配，则返回对应用程序的响应。

#### 总结 ####

1.订阅模式，路由模式，主题模式，他们的相同点就是都使用了交换机，只不过在发送消息给队列时，添加了不同的路由规则。  
2.订阅模式没有路由规则，路由模式为完全匹配规则，主题模式为模糊匹配（正则表达式，完全匹配规则）。  
3.在交换机模式下：队列和路由规则有很大关系，生产者只用关心交换机与路由规则即可，无需关心队列。  
4.消费者不管在什么模式下：永远不用关心交换机和路由规则，消费者永远只关心队列，消费者直接和队列交互。

### 六.Springboot集成RabbitMQ ###

#### 1.生产者搭建 ####

1.引入依赖

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

2.配置文件

spring:
      rabbitmq:
        host: 192.168.3.88
        username: guest
        password: guest
        virtual-host: /
        port: 5672
        publisher-confirms: true #开启确认模式，确保消息的可靠投递，防丢失
        publisher-returns: true #开启回退模式，确保消息的可靠投递，防丢失

3.启动类

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

4.配置类

@Configuration
    public class RabbitmqConfig {
    
        public static final String EXCHANGE_NAME = "boot_topic_exchange";
        public static final String QUEUE_NAME = "boot_queue";
    
        public static final String EXCHANGE_CONFIRM = "confirm_exchange";
        public static final String QUEUE_CONFIRM = "confirm_queue";
    
        public static final String EXCHANGE_TTL = "ttl_exchange";
        public static final String QUEUE_TTL = "ttl_queue";
    
        public static final String EXCHANGE_NORMAL = "normal_exchange";
        public static final String QUEUE_NORMAL = "normal_queue";
    
        public static final String EXCHANGE_DLX = "dlx_exchange";
        public static final String QUEUE_DLX = "dlx_queue";
    
    
        //1.声明交换机
        @Bean("bootExchange")
        public Exchange bootExchange(){
            return ExchangeBuilder.topicExchange(EXCHANGE_NAME)  //声明topic模式的交换机
                    .durable(true)  //开启持久化
                    .build();       //构建交换机
        }
    
        //2.声明队列
        @Bean("bootQueue")
        public Queue bootQueue(){
            return QueueBuilder.durable(QUEUE_NAME).build();
        }
    
        //3.声明交换机和队列的绑定关系(无需注入，只写Bean注解即可)
        @Bean
        public Binding bindExchangeQueue(@Qualifier("bootExchange") Exchange exchange,
                                         @Qualifier("bootQueue") Queue queue){
            return BindingBuilder.bind(queue)   //队列
                    .to(exchange)               //交换机
                    .with("boot.#")  //路由key
                    .noargs();                  //是否还有参数
        }
    
        //------------------------------确认模式
        @Bean("confirmExchange")
        public Exchange confirmExchange(){
            return ExchangeBuilder.directExchange(EXCHANGE_CONFIRM)
                    .durable(true)
                    .build();
        }
        @Bean("confirmQueue")
        public Queue confirmQueue(){
            return QueueBuilder.durable(QUEUE_CONFIRM).build();
        }
        @Bean
        public Binding bindConfirm(@Qualifier("confirmExchange") Exchange exchange,
                                   @Qualifier("confirmQueue") Queue queue){
            return BindingBuilder.bind(queue)   //队列
                    .to(exchange)               //交换机
                    .with("confirm")  //路由key
                    .noargs();                  //是否还有参数
        }
    
        //------------------------------ttl消息过期时间
        @Bean("ttlExchange")
        public Exchange ttlExchange(){
            return ExchangeBuilder.topicExchange(EXCHANGE_TTL)
                    .durable(true)
                    .build();
        }
        @Bean("ttlQueue")
        public Queue ttlQueue(){
            return QueueBuilder.durable(QUEUE_TTL)
                    .withArgument("x-message-ttl",15000) //给队列设置过期时间15s，默认单位是ms
                    .build();
        }
        @Bean
        public Binding bindTtl(@Qualifier("ttlExchange") Exchange exchange,
                                   @Qualifier("ttlQueue") Queue queue){
            return BindingBuilder.bind(queue)   //队列
                    .to(exchange)               //交换机
                    .with("ttl.#")  //路由key
                    .noargs();                  //是否还有参数
        }
    
        //------------------------------死信队列
        //----绑定死信队列的普通队列
        @Bean("normalExchange")
        public Exchange normalExchange(){
            return ExchangeBuilder.topicExchange(EXCHANGE_NORMAL)
                    .durable(true)
                    .build();
        }
        @Bean("normalQueue")
        public Queue normalQueue(){
            return QueueBuilder.durable(QUEUE_NORMAL)
                    .withArgument("x-dead-letter-exchange",EXCHANGE_DLX)//绑定死信交换机
                    .withArgument("x-dead-letter-routing-key","dlx.hehe") //绑定死信队列的路由key
                    .withArgument("x-message-ttl",5000) //设置队列的到期时间
                    .withArgument("x-max-length",10) //设置队列的最大长度
                    .build();
        }
        @Bean
        public Binding bindNormal(@Qualifier("normalExchange") Exchange exchange,
                               @Qualifier("normalQueue") Queue queue){
            return BindingBuilder.bind(queue)   //队列
                    .to(exchange)               //交换机
                    .with("normal.#")  //路由key
                    .noargs();                  //是否还有参数
        }
        //----死信队列（其实就是正常的队列，名字不一样而已）
        @Bean("dlxExchange")
        public Exchange dlxExchange(){
            return ExchangeBuilder.topicExchange(EXCHANGE_DLX)
                    .durable(true)
                    .build();
        }
        @Bean("dlxQueue")
        public Queue dlxQueue(){
            return QueueBuilder.durable(QUEUE_DLX).build();
        }
        @Bean
        public Binding bindDlx(@Qualifier("dlxExchange") Exchange exchange,
                               @Qualifier("dlxQueue") Queue queue){
            return BindingBuilder.bind(queue)   //队列
                    .to(exchange)               //交换机
                    .with("dlx.#")  //路由key
                    .noargs();                  //是否还有参数
        }
    
    }

5.测试类

@RunWith(SpringRunner.class)
    @SpringBootTest
    public class ProducerTest {
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Test
        public void testSend(){
            rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_NAME,
                    "boot.haha",
                    "hello rabbit~~~~~~~");
        }
    
        /**
         * 确认模式，保证消息的可靠投递
         * 步骤：
         *  1.开启确认模式（配置文件中开启）
         *  2.定义回调函数
         *  3.发送消息
         */
        @Test
        public void testConfirm() throws InterruptedException {
            //2.定义回调
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                /**
                 *
                 * @param correlationData  相关配置信息
                 * @param ack   交换机是否成功收到了消息，成功：true  失败：false
                 * @param cause 失败原因
                 */
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    System.out.println("confirm方法被执行了");
                    if (ack){
                        System.out.println("消息接收成功啦！");
                    }else{
                        System.out.println("消息接收失败，失败原因是："+cause);
                    }
                }
            });
            //3.发送消息
            rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_CONFIRM,
                    "confirm",
                    "this is msg for confirm-test");
            Thread.sleep(2000);
        }
    
    
        /**
         * 回退模式，保证消息的可靠投递
         * 步骤：
         *  1.开启回退模式（配置文件中开启）
         *  2.定义回调函数
         *  3.设置交换机处理消息的模式：
         *      1.如果消息没有路由到queue，则消息丢失（默认）
         *      2.如果消息没有路由到queue，则返回消息发送方
         */
        @Test
        public void testReturn() throws InterruptedException {
            //3.设置交换机的处理模式(设置失败后将消息返回消息发送方)
            rabbitTemplate.setMandatory(true);
    
            //2.定义回调
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
                @Override
                public void returnedMessage(Message message,
                                            int replyCode,
                                            String replyText,
                                            String exchange,
                                            String routingkey) {
                    System.out.println("return，接收到发送失败的消息："+message);
                }
            });
            //4.发送消息
            rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_CONFIRM,
                    "confirm111",   //给一个错误的routingKey测试
                    "this is msg for confirm-test");
            Thread.sleep(2000);
        }
    
        @Test
        public void testTtl(){
            rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_TTL,
                    "ttl.hehe",
                    "this is a message for ttl-test");
    
        }
    
        @Test
        public void testDlx(){
            rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_NORMAL,
                    "normal.hehe",
                    "this is a message for dlx-test");
    
        }
    }

#### 2.消费者 ####

1.引入依赖

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

2.配置文件

spring:
      rabbitmq:
        host: 192.168.3.88
        username: guest
        password: guest
        virtual-host: /
        port: 5672
        listener:
          simple:
            acknowledge-mode: manual #消费者使用手动签收（收到消息后的确认方式）
            prefetch: 10 #表示消费端每次拉取10条消息，直到手动确认消费完毕后，才会继续拉去下 一条消息。

3.启动类

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

4.消费者监听类

/**
     * Consumer Ack 消费端收到消息后的确认方式
     * 签收机制：
     *  1.设置手动签收
     *  2.实现ChannelAwareMessageListener接口
     *  3.如果消息处理成功，调用channel的basicAck()签收
     *  4.如果消息处理失败，调用channel的basicNack()拒绝签收，broker重新发送给consumer
     *
     * @author tangbb
     * @date 2021/11/24
     */
    @Component
    public class AckListener implements ChannelAwareMessageListener {
    
        @Override
        @RabbitListener(queues = "confirm_queue")
        public void onMessage(Message message, Channel channel) throws Exception {
            Thread.sleep(1000);
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            try {
                //1.接收转换消息
                System.out.println("AckListener收到的消息为："+new String(message.getBody()));
                //2.处理业务逻辑
                System.out.println("处理业务逻辑");
    //            int i = 3/0;  //制造错误，验证手工签收失败的场景
                //3.手动签收
                channel.basicAck(deliveryTag,true);
            } catch (Exception e) {
                //4.拒绝签收
                System.out.println("拒绝签收啦");
                channel.basicNack(deliveryTag,true,true);
            }
        }
    }

### 七.RabbitMQ之延时队列 ###

#### 1.什么是延时队列 ####

我们通过一些场景来认识延时队列吧：

（1）生成订单30分钟未支付，则自动取消  
（2）生成订单60秒后,给用户发短信  
（3） 滴滴打车订单完成后，如果用户一直不评价，48小时后会将自动评价为5星

#### 2.延时队列和定时任务的区别 ####

1.定时任务有明确的触发时间，延时任务没有  
2.定时任务有执行周期，而延时任务在某事件触发后一段时间内执行，没有执行周期  
3.定时任务一般执行的是批处理操作是多个任务，而延时任务一般是单个任务

#### 3.RabbitMQ怎么实现延时队列 ####

RabbitMQ中没有对消息延迟进行实现，但是我们可以通过TTL以及死信路由来实现消息延迟。

（1）RabbitMQ可以针对Queue和Message设置 x-message-ttl，来控制消息的生存时间，如果超时，则消息变为dead letter。  
（2）RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）两个参数，用来控制队列内出现了deadletter，则按照这两个参数重新路由。

![c32f12cb81b04b459495a17784ed6b3f.png][]

#### 4.TTL(消息过期时间) ####

TTL是RabbitMQ中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间，单位是毫秒。换句话说，如果一条消息设置了TTL属性或者进入了设置TTL属性的队列，那么这条消息如果在TTL设置的时间内没有被消费，则会成为“死信”，如果不设置TTL，表示消息永远不会过期，如果将TTL设置为0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

在创建队列的时候设置队列的“x-message-ttl”属性

/**
    *这样所有被投递到该队列的消息都最多不会存活超过30s
    *如果没有任何处理，消息会被丢弃，如果配置有死信队列，超时的消息会被投递到死信队列
    */
    @Bean
    public Queue taxiOverQueue() {
        Map<String, Object> args = new HashMap<>(2);
        args.put("x-message-ttl", 30000);
        return QueueBuilder.durable(TAXI_OVER_QUEUE).withArguments(args).build();
    }

#### 5.死信队列 ####

**（1）定义**

死信，顾名思义就是无法被消费的消息，某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信。有死信，自然就有了死信队列。

**（2）死信队列的使用场景**

<1>消费者对消息使用了basicReject或者basicNack回复,并且requeue参数设置为false,即不再将该消息重新在消费者间进行投递。  
<2>消息在队列中超时. RabbitMQ可以在单个消息或者队列中设置TTL属性。  
<3>队列中的消息已经超过其设置的最大消息个数。

**（3）死信队列如何使用**

死信交换器不是默认的设置，这里是被投递消息被拒绝后的一个可选行为，是在创建队列的时进行声明的，往往用在对问题消息的诊断上。

死信交换器仍然只是一个普通的交换器，创建时并没有特别要求和操作。在创建队列的时候，声明该交换器将用作保存被拒绝的消息即可，相关的参数是`x-dead-letter-exchange`。

@Bean
    public Queue taxiOverQueue() {
        Map<String, Object> args = new HashMap<>(2);
        // x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", TAXI_DEAD_QUEUE_EXCHANGE);
        // x-dead-letter-routing-key  这里声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key", TAXI_DEAD_KEY);
        return QueueBuilder.durable(TAXI_OVER_QUEUE).withArguments(args).build();
    }

![2593ce2526cf4e35a24e76e4e90a4db7.png][]

**（4）打车超时实现**

在创建队列的时候配置死信交换器并设置队列的“x-message-ttl”属性

这样所有被投递到该队列的消息都最多不会存活超过30s，超时后的消息会被投递到死信交换器

@Bean
    public Queue taxiDeadQueue() {
        return new Queue(TAXI_DEAD_QUEUE,true);
    }
    
    @Bean
    public Queue taxiOverQueue() {
        Map<String, Object> args = new HashMap<>(2);
        // x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", TAXI_DEAD_QUEUE_EXCHANGE);
        // x-dead-letter-routing-key  这里声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key", TAXI_DEAD_KEY);
        // x-message-ttl  声明队列的TTL
        args.put("x-message-ttl", 30000);
        return QueueBuilder.durable(TAXI_OVER_QUEUE).withArguments(args).build();
    }

### 八.RabbitMQ消息可靠性保障 ###

#### 1.RabbiMQ消息处理逻辑 ####

**（1）如果消息到达无人订阅的队列怎么办？**

消息会一直在队列中等待，RabbtiMQ默认队列是无限长度的。

**（2）多个消费者订阅到同一队列怎么办？**

消息以循环的方式发送给消费者，每个消息只会发送给一个消费者

**（3）消息路由到了不存在队列怎么办？**

RabbitMQ默认会直接忽略，即消息丢失了

#### 2.保证RabbiMQ消息可靠性的方式 ####

从下图消息生产投递到消费的过程中，可以从下面三个方面进行保证消息的可靠性

（1）**生产者**

（2）**Broker**（RabbitMQ内部，其实就是RabbtMQ所在的机器）

（3）**消费者**

![16059360231249aeb1caff7146383b6b.png][]

#### 3.生产者保证消息可靠性 ####

（1）**失败通知**

可以启动失败通知，保证消息发布的可靠性

**实现方式**

1.配置文件

spring:
      rabbitmq:
        # 消息在未被队列收到的情况下返回
        publisher-returns: true

2.关键代码，注意需要发送者实现`ReturnCallback`接口方可实现失败通知

/**
      * 失败通知
      * 队列投递错误应答
      * 只有投递队列错误才会应答
      */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        //消息体为空直接返回
        if (null == message) {
            return;
        }
        TaxiBO taxiBO = JSON.parseObject(message.getBody(), TaxiBO.class);
        if (null != taxiBO) {
            //删除rediskey
            redisHelper.handelAccountTaxi(taxiBO.getAccountId());
            //记录错误日志
            recordErrorMessage(taxiBO, replyText, exchange, routingKey, message, replyCode);
        }
    }

失败通知的缺陷：

如果消息正确路由到队列，则发布者不会受到任何通知。带来的问题是无法确保发布消息一定是成功的，因为通知失败的消息可能会丢失。

解决方案：我们可以使用RabbitMQ的发送方确认来实现，它不仅仅在路由失败的时候给我们发送消息，并且能够在消息路由成功的时候也给我们发送消息。

（2）**发送方确认**

一旦消息投递到队列，队列则会向生产者发送一个通知，如果设置了消息持久化到磁盘，则会等待消息持久化到磁盘之后再发送通知。

**实现方式**

1.配置文件

spring:
      rabbitmq:    
        # 开启消息确认机制
        publisher-confirm-type: correlated

2.关键代码，注意需要发送者实现`ConfirmCallback`接口方可实现失败通知

/**
      * 发送方确认
      * 交换器投递后的应答
      * 正常异常都会进行调用
      *
      * @param correlationData
      * @param ack
      * @param cause
      */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        //只有异常的数据才需要处理
        if (!ack) {
            //关联数据为空直接返回
            if (correlationData == null) {
                return;
            }
            //检查返回消息是否为null
            if (null != correlationData.getReturnedMessage()) {
                TaxiBO taxiBO = JSON.parseObject(correlationData.getReturnedMessage().getBody(), TaxiBO.class);
                //处理消息还原用户未打车状态
                redisHelper.handelAccountTaxi(taxiBO.getAccountId());
                //获取交换器
                String exchange = correlationData.getReturnedMessage().getMessageProperties().getHeader("SEND_EXCHANGE");
                //获取队列信息
                String routingKey = correlationData.getReturnedMessage().getMessageProperties().getHeader("SEND_ROUTING_KEY");
                //获取当前的消息体
                Message message = correlationData.getReturnedMessage();
                //记录错误日志
                recordErrorMessage(taxiBO, cause, exchange, routingKey, message, -1);
            }
        }
    }

#### 4.Broker保证消息可靠性 ####

**实现方式**

开启RabbitMQ的持久化，也即消息写入后会持久化到磁盘，此时即使mq挂掉了，重启之后也会自动读取之前存储的额数据。

1.持久化队列

@Bean
    public Queue queue(){
        return new Queue(queueName,true);
    }

2.持久化交换器

@Bean
    DirectExchange directExchange() {
        return new DirectExchange(exchangeName,true,false);
    }

3.发送持久化消息

发送消息时，设置消息的deliveryMode=2
    注意：如果使用SpringBoot的话，发送消息时自动设置deliveryMode=2，不需要人工再去设置

4.broker持久化总结优化

假如消息到达队列之后，还未保存到磁盘mq就挂掉了，此时还是有很小的几率会导致消息丢失的。

这就要mq的持久化和前面的confirm进行配合使用，只有当消息写入磁盘后才返回ack，那么就是在持久化之前mq挂掉了，但是由于生产者没有接收到ack信号，此时可以进行消息重发。

#### 5.消费者保证消息可靠性 ####

**实现方式：消费者手动确认**

RabbitMQ默认是自动ack的，此时需要将其修改为手动ack，也即**自己的程序确定消息已经处理完成后，手动提交ack**，此时如果再遇到消息未处理进程就挂掉的情况，由于没有提交ack，RabbitMQ就不会删除这条消息，而是会把这条消息发送给其他消费者处理，但是消息是不会丢的。

1.配置文件

spring:
      rabbitmq:
        listener:
          simple:
            acknowledge-mode: manual  # 表示开启手动ack，该配置项的其他两个值分别是none和auto

**auto**：消费者根据程序执行正常或者抛出异常来决定是提交ack或者nack，不要把none和auto搞混了

**manual**: 手动ack，用户必须手动提交ack或者nack

**none**: 没有ack机制

2.消费者关键代码

@RabbitListener(
                bindings =
                        {
                                @QueueBinding(value = @Queue(value = RabbitConfig.TAXI_DEAD_QUEUE, durable = "true"),
                                        exchange = @Exchange(value = RabbitConfig.TAXI_DEAD_QUEUE_EXCHANGE), key = RabbitConfig.TAXI_DEAD_KEY)
                        })
        @RabbitHandler
        public void processOrder(Message massage, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
            TaxiBO taxiBO = JSON.parseObject(massage.getBody(), TaxiBO.class);
            try {
                //开始处理订单
                logger.info("处理超时订单，订单详细信息：" + taxiBO.toString());
                taxiService.taxiTimeout(taxiBO);
                //手动确认机制
                channel.basicAck(tag, false);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

### 九.RabbitMQ业务可靠性分析 ###

![ec61f5ace9be44a3bc8f330bf13a0af9.png][]

在这个业务场景中，用户发起打车请求，如果用户消息丢失，对整体业务是没有任何影响的，用户可以再次发起打车操作，这个消息丢失问题概率很低，可以进行简单化设计，如果出现发送失败只需要回退redis中的操作即可。

#### 幂等性校验 ####

因为使用了延时队列，对于这个业务来说是不需要进行幂等性校验的，因为第一次超时时如果存在redis用户排名的key就会被删除，下一次redis没有的值在删除一次，这种操作是幂等的，所以不需要考虑幂等性

#### 数据回滚 ####

虽然无需做到消息完全不丢失以及消息的幂等性，但是需要考虑如果出现问题，需要将插入Redis的的key值回滚掉，防止影响业务正常判断

注意：

rabbitmq的默认端口为5672，界面默认端口为15672，java程序连接mq端口使用5672即可

rabbitmq的端口在ebin下的rabbitmq.app中配置

[https_blog.csdn.net_qq_45443475_article_details_119934030_spm_1001.2014.3001.5502]: https://blog.csdn.net/qq_45443475/article/details/119934030?spm=1001.2014.3001.5502
[f81344fa6d244c1892bf22acd9d84812.png]: https://img-blog.csdnimg.cn/f81344fa6d244c1892bf22acd9d84812.png
[64280aed55184213b8cfd6ad309d8f49.png]: https://img-blog.csdnimg.cn/64280aed55184213b8cfd6ad309d8f49.png
[c7e3048d27c44ae7ae71fbf3d761c2f1.png]: https://img-blog.csdnimg.cn/c7e3048d27c44ae7ae71fbf3d761c2f1.png
[59921272f5f24e0098dfd801cd49e038.png]: https://img-blog.csdnimg.cn/59921272f5f24e0098dfd801cd49e038.png
[151c20fec79943cd94adc4be0ca6955b.png]: https://img-blog.csdnimg.cn/151c20fec79943cd94adc4be0ca6955b.png
[bf6a1333d867423e9d58654cc2e5b85c.png]: https://img-blog.csdnimg.cn/bf6a1333d867423e9d58654cc2e5b85c.png
[0a4dd7df68964dd588521b7ebf781455.png]: https://img-blog.csdnimg.cn/0a4dd7df68964dd588521b7ebf781455.png
[bf3b46a7871c439a80922c214b37c453.png]: https://img-blog.csdnimg.cn/bf3b46a7871c439a80922c214b37c453.png
[f7c170efe1ec4ab8a70560fe47c8a21b.png]: https://img-blog.csdnimg.cn/f7c170efe1ec4ab8a70560fe47c8a21b.png
[c32f12cb81b04b459495a17784ed6b3f.png]: https://img-blog.csdnimg.cn/c32f12cb81b04b459495a17784ed6b3f.png
[2593ce2526cf4e35a24e76e4e90a4db7.png]: https://img-blog.csdnimg.cn/2593ce2526cf4e35a24e76e4e90a4db7.png
[16059360231249aeb1caff7146383b6b.png]: https://img-blog.csdnimg.cn/16059360231249aeb1caff7146383b6b.png
[ec61f5ace9be44a3bc8f330bf13a0af9.png]: https://img-blog.csdnimg.cn/ec61f5ace9be44a3bc8f330bf13a0af9.png