RabbitMQ消息队列工作原理及集成使用

心已赠人 2023-05-28 15:55 2阅读 0赞

**消息 **指的是两个应用间传递的数据。数据的类型有很多种形式，可能只包含文本字符串，也可能包含嵌入对象。

**消息队列(Message Queue)** 是在消息的传输过程中保存消息的容器。

在消息队列中，通常有生产者和消费者两个角色。生产者只负责发送数据到消息队列，谁从消息队列中取出数据处理他不管。消费者只负责从消息队列中取出数据处理，他不管这是谁发送的数据。

# 一、应用场景 #

## 1.1  系统间解耦  ##

我们遇到过这样的需求：一个商品销售系统，用户下单后，订单系统需要通知库存系统；或者一个运营后台系统新建单位或者用户时，通知对应的业务平台初始化该单位/用户的配置。遇到类似这两种需求的时候，传统的做法是订单系统调用库存系统的接口或者运营系统调用业务平台的接口，这种“直调”的方式完成功能的同时还有一些缺陷：

*  被调用方接口一旦失败，调用方也将失败。例如库存系统无法访问，则订单减库存也将失败，从而导致订单失败
 *  被调用方系统与调用方系统耦合

这时候可以引入消息队列作为第三方媒介，用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功；库存系统订阅下单的消息，采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作。加上消息队列的好处就是在下单时库存系统不能正常使用，也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦。

（ps：消息队列和RPC的区别与比较：RPC: 异步调用，及时获得调用结果，具有强一致性结果，关心业务调用处理结果。消息队列：两次异步RPC调用，将调用内容在队列中进行转储，并选择合适的时机进行投递（错峰流控））

## 1.2  异步提升效率 ##

我们也会遇到这样的场景：系统注册新的用户后，给这个用户发确认邮件或者短信，所有都完成了返回一个注册成功的消息。无论是串行实现还是并行实现，都会影响效率，因为发短信和邮件并不是主体业务功能，真正的主体业务还要等一些非必要的功能运行完才能最终实现。这时可以引入消息对列，邮件和信息订阅对应的队列消息，进行异步处理。

## 1.3  流量削峰 ##

我们也会遇到这样的场景：高并发流量大的场景，例如整点秒杀、大型直播。

大流量时MySQL可能忙不过来，比如每秒MySQL只能处理6000个请求，但是现在一秒钟来了1万个请求，服务器可能就宕机了，那就可以将过多的处理先放到消息队列（消息队列可以应付的过来），然后系统再慢慢地从队列中按照顺序消费这些信息，直到消费完所有消息。在我们秒杀抢购商品的时候，系统会提醒我们稍等排队中，而不是像几年前一样页面卡死或报错给用户。

![6ef4a1a08153ea7f2b032853770e1bb2.png][]

## 1.4  统一通信标准 ##

不同进程（process）之间传递消息时，可以通过消息队列实现标准化，将消息的格式规范化。

说了这么久都在讲Mq的好处，实际上它也有槽点：

*  系统的可用性降低  
    系统引入的外部依赖越多，系统越容易挂掉，本来只是A系统调用BCD三个系统接口就好，ABCD四个系统不报错整个系统会正常运行。引入了MQ之后，虽然ABCD系统没出错，但MQ挂了以后，整个系统也会崩溃。
 *  
 *  系统的复杂性提高  
    引入了MQ之后，需要考虑的问题也变得多了，如何保证消息没有重复消费？如何保证消息不丢失？怎么保证消息传递的顺序？
 *  一致性问题  
    A系统发送完消息直接返回成功，但是BCD系统之中若有系统写库失败，则会产生数据不一致的问题。

消息队列是一种十分复杂的架构，引入它有很多好处，但是也得针对它带来的坏处做各种额外的技术方案和架构来规避。引入MQ系统复杂度提升了一个数量级，但是在有些场景下，就是复杂十倍百倍，还是需要使用MQ。

# 二、什么是消息队列？ #

消息队列（Message Queue，简称MQ），从字面意思上看，本质是个队列，FIFO先入先出，只不过队列中存放的内容是message而已。详细的队列数据结构可以查看下面的文章：

[数据结构--队列Queue的实现原理\_沙滩的流沙520的博客-CSDN博客][--_Queue_520_-CSDN]

MQ框架非常之多，比较流行的有RabbitMq、ActiveMq、ZeroMq、kafka，以及阿里开源的RocketMQ。本文主要介绍RabbitMq。下面是各MQ框架技术选型对比：

Kafka的吞吐量高达17.3w/s，不愧是高吞吐量消息中间件的行业老大。这主要取决于它的队列模式保证了写磁盘的过程是线性IO。此时broker磁盘IO已达瓶颈。

RocketMQ也表现不俗，吞吐量在11.6w/s，磁盘IO %util已接近100%。RocketMQ的消息写入内存后即返回ack，由单独的线程专门做刷盘的操作，所有的消息均是顺序写文件。

RabbitMQ的吞吐量5.95w/s，CPU资源消耗较高。它支持AMQP协议，实现非常重量级，为了保证消息的可靠性在吞吐量上做了取舍。我们还做了RabbitMQ在消息持久化场景下的性能测试，吞吐量在2.6w/s左右。

在服务端处理同步发送的性能上，Kafka>RocketMQ>RabbitMQ。

对比了最简单的小消息发送场景,Kafka暂时胜出。但是,作为经受过历次双十一洗礼的RocketMQ,在互联网应用场景中更有它优越的一面。

# RabbitMQ #

是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正是如此，使的它变的非常重量级，更适合于企业级的开发。

**吞吐量：**

*  rabbitMQ在吞吐量方面稍逊于kafka，他们的出发点不一样，rabbitMQ支持对消息的可靠的传递，支持事务，不支持批量的操作；基于存储的可靠性的要求存储可以采用内存或者硬盘。
 *  kafka具有高的吞吐量，内部采用消息的批量处理，zero-copy机制，数据的存储和获取是本地磁盘顺序批量操作，具有O(1)的复杂度，消息处理的效率很高。

Kafka是LinkedIn开源的分布式发布-订阅消息系统。Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输。0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务。

RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。AMQP协议更多用在企业系统内，对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。

RocketMQ是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka，但并不是Kafka的一个Copy，它对消息的可靠传输及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。

# 三、RabbitMQ #

## 3.1  介绍 ##

RabbitMQ是一个实现了AMQP（Advanced Message Queuing Protocol）高级消息队列协议的消息队列服务，Erlang语言编写。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。 AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面有很好表现。

## 3.2  RabbitMQ中的重要术语 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70][]

*  **ConnectionFactory（连接管理器）：**应用程序与Rabbit之间建立连接的管理器；
 *  **Channel（信道）：**消息推送使用的通道；
 *  **Exchange（交换器）：**用于接受、分配消息。生产者将消息发送到交换器，由交换器将消息路由到一个或者多个队列中。当路由不到时，或返回给生产者或直接丢弃；
 *  **Queue（队列）**：RabbitMQ的内部对象，用于存储生产者的消息；一个queue中可以存放的消息数量可以人为是无限制的，限制因素是机器内存，但是也不能存储过多消息，会降低处理效率。
 *  **RoutingKey（路由键）**：用于把生成者的数据分配到交换器上。生产者将消息发送给交换器的时候，会指定一个RoutingKey，用来指定这个消息的路由规则，这个RoutingKey需要与交换器类型和绑定键(BindingKey)联合使用才能最终生效；
 *  **BindingKey（绑定键）**：用于把交换器的消息绑定到队列上。通过绑定将交换器和队列关联起来，一般会指定一个BindingKey，这样RabbitMq就知道如何正确路由消息到队列了。
 *  **Virtual Hosts（虚拟主机****）**：每一个RabbitMQ服务器都能创建虚拟的消息服务器，我们称之为虚拟主机（virtual host），简称为vhost。每一个vhost本质上是一个独立的小型RabbitMQ服务器，拥有自己独立的队列、交换器以及绑定关系等待，并且它拥有自己独立的权限。vhost就像是虚拟机与物理服务器一样，它们在各个实例间提供逻辑上的分离，允许你为不同程序安全保密地运行数据，它既能将同一个RabbitMQ的中众多客户区分开，又可以避免队列和交换器等命令冲突。vhost之间是绝对隔离的，你无法将vhost1中的交换器与vhost2中的队列进行绑定，这样既保证了安全性，又可以确保可移植性。如果在使用RabbitMQ达到一定规模的时候，用户可以对业务功能、场景进行归类区分，并为之分配独立的vhost。
 *  **Broker（**节点**）：**可以看做RabbitMQ的服务节点。或者看成消息队列服务进程，此进程包括两个部分：Exchange和Queue。

**RoutingKey和BindingKey最大长度是 255 字节。**

### 3.2.1  Exchange ###

生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

![format_png][]

### 3.2.2  routing key ###

消息可以认为由payload+ routing key组成，payload就是消息内容数据，routing key来指定这个消息的路由规则。这个routing key需要与Exchange Type及binding key联合使用才能最终生效。 在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。 RabbitMQ为routing key设定的长度限制为255 bytes。

### 3.2.3  Binding key ###

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；**当 binding key 与 routing key 相匹配时**，消息将会被路由到对应的Queue中。在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。 binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

![format_png 1][]

上面的各部分组成了RabbitMQ的整个流程：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70 1][]

# 3.3  RabbitMQ运转流程 #

消息生产者，就是投递消息的一方。

消息一般包含两个部分：消息体（payload)和标签(Label)。

消费消息，也就是接收消息的一方。

消费者连接到RabbitMQ服务器，并订阅到队列上。消费消息时只消费消息体，丢弃标签。

### 生产者发送消息 ###

1、生产者连接到 RabbitMQ Broker，建立一个连接（Connection），开启一个信道（Channel）

2、生产者声明一个交换器，并设置相关属性，比如交换机类型、是否持久化等

3、生产者声明一个队列并设置相关属性，比如是否排他、是否持久化、是否自动删除等

4、生产者通过路由键将交换器和队列绑定起来

5、生产者发送消息至 RabbitMQ Broker，其中包含路由键、交换器等信息

6、相应的交换器根据收到的路由键查找相匹配的队列

7、如果找到，则将从生产者发送过来的消息存入相应的队列中

8、如果没有找到，则根据生产者配置的属性选择丢弃还是回退给生产者

9、关闭信道

10、关闭连接

### 消费者接收消息的过程 ###

1、消费者连接到 RabbitMQ Broker，建立一个连接（Connection），开启一个信道（Channel）

2、消费者向 RabbitMQ Broker 请求消费相应队列中的消息，可能会设置相应的回调函数，以及做一些准备工作

3、等待 RabbitMQ Broker 回应并投递相应队列中的消息，消费者接收消息

4、消费者确认（ack）接收到的消息

5、RabbitMQ 从队列中删除相应已经被确认的消息

6、关闭信道

7、关闭连接

从概念上来说，消息路由必须有三部分：**交换器、路由、绑定**。生产者把消息发布到交换器上；绑定决定了消息如何从路由器路由到特定的队列；消息最终到达队列，并被消费者接收。

消息发布到交换器时，消息将拥有一个路由键（routing key），在消息创建时设定。  
通过队列路由键，可以把队列绑定到交换器上。  
消息到达交换器后，RabbitMQ会将消息的路由键与队列的路由键进行匹配（针对不同的交换器有不同的路由规则）。如果能够匹配到队列，则消息会投递到相应队列中；如果不能匹配到任何队列，消息将进入 “黑洞”。

## 3.4  RabbitMQ 重要对象 ##

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。

1.  Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑
2.  ConnectionFactory为Connection的制造工厂
3.  大部分的业务操作在Channel这个接口中完成，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等
4.  Queue是RabbitMQ的内部对象，消息都只能存储在Queue中，生产者生产消息并最终投递到Queue中，消费者从Queue中获取消息并消费。

## 3.5  RabbitMQ中的重要机制 ##

### **3.5.1  Message acknowledgment（消息回执****）** ###

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。 这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻。另外pub message是没有ack的。

### 3.5.2  Message durability（消息持久化） ###

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。

### 3.5.3  Prefetch count（预载数量） ###

如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

### 3.5.4  RPC（远程过程调用） ###

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。 但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

[RabbitMQ][]  中实现 RPC 的机制是：

*  客户端发送请求（消息）时，在消息的属性（`MessageProperties` ，在`AMQP` 协议中定义了14中`properties` ，这些属性会随着消息一起发送）中设置两个值`replyTo`（一个`Queue` 名称，用于告诉服务器处理完成后将通知我的消息发送到这个`Queue`中）和`correlationId` （此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
 *  服务器端收到消息并处理
 *  服务器端处理完消息后，将生成一条应答消息到`replyTo` 指定的`Queue` ，同时带上`correlationId` 属性
 *  客户端之前已订阅`replyTo` 指定的`Queue` ，从中收到服务器的应答消息后，根据其中的`correlationId` 属性分析哪条请求被执行了，根据执行结果进行后续业务处理

## 3.6  Exchange Types ##

RabbitMQ常用的Exchange Type有fanout、direct、topic这3种。

### 3.6.1  direct ###

binding key与routing key **完全匹配 **的才路由。

![format_png 2][]

以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

### 3.6.2  topic ###

direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

*  routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
 *  binding key与routing key一样也是句点号“. ”分隔的字符串
 *  binding key中可以存在两种特殊字符“\*”与“\#”，用于做模糊匹配，其中“\*”用于匹配一个单词，“\#”用于匹配多个单词（可以是零个）

![format_png 3][]

以上图中的配置为例：

*  routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2
 *  routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2
 *  routingKey=”lazy.brown.fox”的消息会路由到Q2
 *  routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）
 *  routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将**会被丢弃**，因为它们没有匹配任何bindingKey。

### 3.6.3 fanout ###

fanout类型的Exchange会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

生产者发送到Exchange的所有消息都会路由到图中的两个Queue，并最终被两个消费者消费。

### 3.6.4 Headers Exchange ###

这种交换机用的相对没这么多。**它跟上面三种有点区别，它的路由不是用routingKey进行路由匹配，而是在匹配请求头中所带的键值进行路由**。

创建队列需要设置绑定的头部信息，有两种模式：**全部匹配和部分匹配**。

![d2d51d71282ba3536180a0d2b47f3f6a.png][]

![a3982540411b4afeb7a65088dc09ea3f.png][]

![980910fc84bcc0de248d3f2f45b94e8a.png][]

测试全匹配的队列A：

# ![20f7c46a497ccda6d84cb65cb2440bed.png][] #

# ![e53c3a3c296337ee5a88bbc974fc8ef3.png][] #

再测试部分匹配的队列B：

![761b452420b966d9681750bf72ffc12b.png][]

![300c65e8d2f0c96ef4e0ce0ed2db1959.png][]

# 四、Spring boot 集成 RabbitMQ #

## 4.1  引入maven依赖 ##

![20200317093054938.png][]

这里是我们私服的依赖

当然服务要连接到RabbitMQ的服务上：

spring:
        rabbitmq:
            host: 127.0.0.1
            port: 5672
            username: guest
            password: guest

## 4.2  生产消息 ##

@RestController
    @Slf4j
    public class DirectPublish {
    
            @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @GetMapping("/userAdd")
        public String publish() {
            // 要发送的数据
            User user = getUser();
            // 交换机类型是fanout，路由到所有队列，因此不需要指定routing key
            rabbitTemplate.convertAndSend(FanoutRabbitConfig.GOV_UC_USER_ADD_EXCHANGE, null, user);
            log.info(user.toString());
            return "ok";
        }
    
        private User getUser() {
            Random random = new Random();
            return User.builder()
                    .id("-6666666")
                    .name("葛锋" + random.nextInt(100))
                    .loginName("zk")
                    .orgAccountId("670869647114347")
                    .build();
        }
    }

## 4.3  消费消息 ##

@RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "GOV_THIRDPARTY", durable = "true"),
            exchange = @Exchange(name = "GOV_THIRDPARTY_EXCHANGE"),
            key = "GOV_THIRDPARTY_SEEYOU"))
    @Component
    public class listener {
        @RabbitHandler
        public void handle(@Payload User user, AmqpMessageHeaderAccessor messageHeaderAccessor, Channel channel) throws IOException {
            System.out.println("消费队列===>" + user.toString());
        }
    }

Rabbit的集成使用还是很简单的，简单注解即可。

这里RabbitTemplate已经封装了ConnectionFactory、Channel等。

真实的过程如下：

生产消息：

public class Send {
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    
            // 消息内容
            String message = user.toString();
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }

消费消息：

public class Recv {
    
        private final static String QUEUE_NAME = "queue_work1";
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println("消息内容： " + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

# 六、死信队列 #

为了保证业务的消息数据不丢失，需要使用到RabbitMQ的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。

## 6.1 什么是死信队列？ ##

死信（Dead Letter），“是RabbitMQ中的一种消息机制，当你在消费消息时，如果队列里的消息出现以下情况，该消息将会成为“死信”：

*  消息在队列的存活时间超过设置的TTL时间。
 *  消息队列的消息数量已经超过最大队列长度。
 *  消息被否定确认，使用 `channel.basicNack` 或 `channel.basicReject` ，并且此时`requeue` 属性被设置为`false。`

“死信”消息会被RabbitMQ进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中，如果没有配置，则该消息将会被丢弃。

## 6.2 如何配置死信队列？ ##

大概可以分为以下步骤：

1.  配置业务队列，绑定到业务交换机上
2.  为业务队列配置死信交换机和路由key
3.  为死信交换机配置死信队列

并不是直接声明一个公共的死信队列，然后保存所有死信消息。而是为每个需要使用死信的业务队列配置一个死信交换机（这里同一个项目的死信交换机可以共用一个），然后为每个业务队列分配一个单独的路由key。有了死信交换机和路由key后，接下来，就像配置业务队列一样，配置死信队列，然后绑定在死信交换机上。

所以说，死信队列并不是什么特殊的队列，只不过是绑定在死信交换机上的队列。死信交换机也不是什么特殊的交换机，只不过是用来接受死信的交换机，可以为任何类型【Direct、Fanout、Topic】。

# 七、延时队列 #

要实现延时队列，往往还要结合死信队列。

## 7.1 死信队列 ##

### 7.1.1  什么是死信队列？ ###

死信（Dead Letter），是RabbitMQ中的一种消息机制，当你在消费消息时，如果队列里的消息出现以下情况：

1.  消息被否定确认，使用 `channel.basicNack` 或 `channel.basicReject` ，并且此时`requeue` 属性被设置为`false`。
2.  消息在队列的存活时间超过设置的TTL时间。
3.  消息队列的消息数量已经超过最大队列长度。

那么该消息将成为“死信”。

“死信”消息会被RabbitMQ进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中；如果没有配置，则该消息将会被丢弃。

### 7.1.2 死信消息生命周期 ###

1.  业务消息被投入业务队列
2.  消费者消费业务队列的消息，由于处理过程中发生异常，于是进行了nck或者reject操作
3.  被nck或reject的消息由RabbitMQ投递到死信交换机中
4.  死信交换机将消息投入相应的死信队列
5.  死信队列的消费者消费死信消息

### 7.1.3 如何配置死信队列？ ###

大概可以分为以下步骤：

1.  配置业务队列，绑定到业务交换机上
2.  为业务队列配置死信交换机和路由key
3.  为死信交换机配置死信队列

并不是直接声明一个公共的死信队列，然后把所有死信消息放入这个公共死信队列。而是为每个需要使用死信的业务队列配置一个死信交换机，这里同一个项目的死信交换机可以共用一个，然后为每个业务队列分配一个单独的路由key。有了死信交换机和路由key后，接下来，就像配置业务队列一样，配置死信队列，然后绑定在死信交换机上。

死信队列并不是什么特殊的队列，只不过是绑定在死信交换机上的队列。死信交换机也不是什么特殊的交换机，只不过是用来接受死信的交换机，所以可以为任何类型【Direct、Fanout、Topic】。

一般来说，会为每个业务队列分配一个独有的路由key，并对应的配置一个死信队列进行监听，也就是说，一般会为每个重要的业务队列配置一个死信队列。

（1）创建一个Config类：

package com.example.pp.demos.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * Feng, Ge 2020/8/15 20:21
     */
    @Configuration
    public class RabbitMQConfig {
        public static final String BUSINESS_EXCHANGE_NAME = "dead.letter.demo.simple.business.exchange";
        public static final String BUSINESS_QUEUEA_NAME = "dead.letter.demo.simple.business.queuea";
        public static final String BUSINESS_QUEUEB_NAME = "dead.letter.demo.simple.business.queueb";
        public static final String DEAD_LETTER_EXCHANGE = "dead.letter.demo.simple.deadletter.exchange";
        public static final String DEAD_LETTER_QUEUEA_ROUTING_KEY = "dead.letter.demo.simple.deadletter.queuea.routingkey";
        public static final String DEAD_LETTER_QUEUEB_ROUTING_KEY = "dead.letter.demo.simple.deadletter.queueb.routingkey";
        public static final String DEAD_LETTER_QUEUEA_NAME = "dead.letter.demo.simple.deadletter.queuea";
        public static final String DEAD_LETTER_QUEUEB_NAME = "dead.letter.demo.simple.deadletter.queueb";
    
        // 声明业务Exchange
        @Bean("businessExchange")
        public FanoutExchange businessExchange(){
            return new FanoutExchange(BUSINESS_EXCHANGE_NAME);
        }
    
        // 声明死信Exchange
        @Bean("deadLetterExchange")
        public DirectExchange deadLetterExchange(){
            return new DirectExchange(DEAD_LETTER_EXCHANGE);
        }
    
        // 声明业务队列A
        @Bean("businessQueueA")
        public Queue businessQueueA(){
            Map<String, Object> args = new HashMap<>(2);
    //       x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
            args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
    //       x-dead-letter-routing-key  这里声明当前队列的死信路由key
            args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEA_ROUTING_KEY);
            return QueueBuilder.durable(BUSINESS_QUEUEA_NAME).withArguments(args).build();
        }
    
        // 声明业务队列B
        @Bean("businessQueueB")
        public Queue businessQueueB(){
            Map<String, Object> args = new HashMap<>(2);
    //       x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
            args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
    //       x-dead-letter-routing-key  这里声明当前队列的死信路由key
            args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEB_ROUTING_KEY);
            return QueueBuilder.durable(BUSINESS_QUEUEB_NAME).withArguments(args).build();
        }
    
        // 声明死信队列A
        @Bean("deadLetterQueueA")
        public Queue deadLetterQueueA(){
            return new Queue(DEAD_LETTER_QUEUEA_NAME);
        }
    
        // 声明死信队列B
        @Bean("deadLetterQueueB")
        public Queue deadLetterQueueB(){
            return new Queue(DEAD_LETTER_QUEUEB_NAME);
        }
    
        // 声明业务队列A绑定关系
        @Bean
        public Binding businessBindingA(@Qualifier("businessQueueA") Queue queue,
                                        @Qualifier("businessExchange") FanoutExchange exchange){
            return BindingBuilder.bind(queue).to(exchange);
        }
    
        // 声明业务队列B绑定关系
        @Bean
        public Binding businessBindingB(@Qualifier("businessQueueB") Queue queue,
                                        @Qualifier("businessExchange") FanoutExchange exchange){
            return BindingBuilder.bind(queue).to(exchange);
        }
    
        // 声明死信队列A绑定关系
        @Bean
        public Binding deadLetterBindingA(@Qualifier("deadLetterQueueA") Queue queue,
                                          @Qualifier("deadLetterExchange") DirectExchange exchange){
            return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEA_ROUTING_KEY);
        }
    
        // 声明死信队列B绑定关系
        @Bean
        public Binding deadLetterBindingB(@Qualifier("deadLetterQueueB") Queue queue,
                                          @Qualifier("deadLetterExchange") DirectExchange exchange){
            return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEB_ROUTING_KEY);
        }
    }

这里声明了两个Exchange，一个是业务Exchange，另一个是死信Exchange，业务Exchange下绑定了两个业务队列，业务队列都配置了同一个死信Exchange，并分别配置了路由key，在死信Exchange下绑定了两个死信队列，设置的路由key分别为业务队列里配置的路由key。

（2）配置文件application.yml

spring:
      rabbitmq:
        host: localhost
        password: guest
        username: guest
        listener:
          type: simple
          simple:
              default-requeue-rejected: false
              acknowledge-mode: manual

（3）业务队列的消费代码：

@Component
    @Slf4j
    public class BusinessMessageReceiver {
        public static final String BUSINESS_QUEUEA_NAME = "dead.letter.demo.simple.business.queuea";
        public static final String BUSINESS_QUEUEB_NAME = "dead.letter.demo.simple.business.queueb";
    
        @RabbitListener(queues = BUSINESS_QUEUEA_NAME)
        public void receiveA(Message message, Channel channel) throws IOException {
            String msg = new String(message.getBody());
            log.info("收到业务消息A：{}", msg);
            boolean ack = true;
            Exception exception = null;
            try {
                if (msg.contains("deadletter")){
                    throw new RuntimeException("dead letter exception");
                }
            } catch (Exception e){
                ack = false;
                exception = e;
            }
            if (!ack){
                log.error("消息消费发生异常，error msg:{}", exception.getMessage(), exception);
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            } else {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        }
    
        @RabbitListener(queues = BUSINESS_QUEUEB_NAME)
        public void receiveB(Message message, Channel channel) throws IOException {
            String msg = new String(message.getBody());
            log.info("收到业务消息B：{}", msg);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

死信队列的消费者：

@Component
    @Slf4j
    public class DeadLetterMessageReceiver {
    
        public static final String DEAD_LETTER_QUEUEA_NAME = "dead.letter.demo.simple.deadletter.queuea";
        public static final String DEAD_LETTER_QUEUEB_NAME = "dead.letter.demo.simple.deadletter.queueb";
    
        @RabbitListener(queues = DEAD_LETTER_QUEUEA_NAME)
        public void receiveA(Message message, Channel channel) throws IOException {
            System.out.println("收到死信消息A：" + new String(message.getBody()));
            log.info("死信消息properties：{}", message.getMessageProperties());
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    
        @RabbitListener(queues = DEAD_LETTER_QUEUEB_NAME)
        public void receiveB(Message message, Channel channel) throws IOException {
            System.out.println("收到死信消息B：" + new String(message.getBody()));
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

消息生产者，并通过controller层来生产消息：

@Component
    public class BusinessMessageSender {
    
        public static final String BUSINESS_EXCHANGE_NAME = "dead.letter.demo.simple.business.exchange";
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendMsg(String msg){
            rabbitTemplate.convertSendAndReceive(BUSINESS_EXCHANGE_NAME, "", msg);
        }
    }

@RequestMapping("rabbitmqTest")
    @RestController
    public class RabbitMQMsgController {
        @Autowired
        private BusinessMessageSender sender;
    
        @RequestMapping("sendMsg")
        public void sendMsg(String msg){
            sender.sendMsg(msg);
        }
    }

测试访问：

http://localhost:8080/rabbitmqTest/sendMsg?msg=msg

结果：

![20200815211455810.png][]

两个Consumer都正常收到了消息，这代表正常消费的消息，ack后正常返回。再测试nck的消息：

http://localhost:8080/rabbitmqTest/sendMsg?msg=deadletter

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70 2][]

消息被NCK后，会抛到死信队列中，被死信队列的消费者消费。

### 7.1.4 死信队列应用场景 ###

一般用在较为重要的业务队列中，确保未被正确消费的消息不被丢弃，一般发生消费异常可能原因主要有由于消息信息本身存在错误导致处理异常，处理过程中参数校验异常，或者因网络波动导致的查询异常等等，当发生异常时，避免每次通过日志来获取原消息，然后让运维帮忙重新投递消息。

通过配置死信队列，可以让未正确处理的消息暂存到另一个队列中，待后续排查清楚问题后，编写相应的处理代码来处理死信消息，这样比手工恢复数据的效率和质量要好很多。

## 7.2 RabbitMQ延迟队列 ##

### 7.2.1  什么是RabbitMQ延迟队列？ ###

首先，它是一种队列，队列意味着内部的元素是`有序`的，元素出队和入队是有方向性的，元素从一端进入，从另一端取出。

其次，最重要的特性就体现在它的延时属性上，普通的队的元素是希望被早点取出处理，**延时队列的元素是希望在指定的时间得到取出和处理，所以延时队列中的元素是都是带时间属性的**，通常来说是需要被处理的消息或者任务。

简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

### 7.2.2  延迟队列的使用场景 ###

1.  按照配置的推送规则，提前一个小时发送推送消息。
2.  订单在十分钟之内未支付则自动取消。
3.  新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
4.  账单在一周内未支付，则自动结算。
5.  用户注册成功后，如果三天内没有登陆则进行短信提醒。
6.  用户发起退款，如果三天内没有得到处理则通知相关运营人员。
7.  预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。

这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务。

看起来似乎使用定时任务，一直轮询数据，每秒/分钟/小时查一次，取出需要被处理的数据，然后处理不就完事了吗？如果数据量比较少，确实可以这样做，但对于数据量比较大，并且时效性较强的场景仍旧使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单的检查，同时会给数据库带来很大压力，无法满足业务要求而且性能低下。这时可以考虑使用延时队列。

###  7.2.3  TTL ###

TTL（Time To Live）是RabbitMQ中的一个高级特性。

TTL是RabbitMQ中一个消息或者队列的属性，表明`一条消息或者该队列中的所有消息的最大存活时间`，单位是毫秒。

若一条消息设置了TTL属性或进入了设置TTL属性的队列，那么这条消息如果在TTL设置的时间内没有被消费，则会成为“死信”。

上面这句话也隐含了TTL的设置有两种设置方式：队列的TTL和消息的TTL。如果同时配置了队列的TTL和消息的TTL，那么较小的那个值将会被使用。

**TTL设置方式：**

方式一：创建队列的时候设置队列的“x-message-ttl”属性：

Map<String, Object> args = new HashMap<String, Object>();
    args.put("x-message-ttl", 6000);
    channel.queueDeclare(queueName, durable, exclusive, autoDelete, args);

所有被投递到该队列的消息都最多不会存活超过6s，这种配置好处是可以统一配置TTL的时间，所有的消息一样的TTL。

方式二：针对每条消息设置TTL：

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
    builder.expiration("6000");
    AMQP.BasicProperties properties = builder.build();
    channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg body".getBytes());

这种就很灵活，每个消息有自己的TTL。

但这两种方式是有区别的，**如果设置了队列的TTL属性，那么一旦消息过期，就会被队列丢弃，而第二种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间。**

另外，还需要注意的一点是，如果不设置TTL，表示消息永远不会过期，如果将TTL设置为0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

###  7.2.4  RabbitMQ的延时队列实现 ###

延时队列可以利用  **TTL + 死信队列** 实现。

延时队列想要消息延迟一段时间再被处理，TTL能让消息在延迟一段时间之后成为死信，消费者消费死信队列里的消息即可。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70 3][]

生产者生产一条延时消息，根据需要延时时间的不同，利用不同的routingkey将消息路由到不同的延时队列，每个队列都设置了不同的TTL属性，并绑定在同一个死信交换机中，消息过期后，根据routingkey的不同，又会被路由到不同的死信队列中，消费者只需要监听对应的死信队列进行处理即可。

### 7.2.5  RabbitMQ的延时队列代码实例 ###

如果使用在消息属性上设置TTL的方式，消息可能并不会按时“死亡“，因为RabbitMQ只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，则第二个消息并不会优先得到执行。因此需要一个插件解决这个问题，[Community Plugins — RabbitMQ][Community Plugins _ RabbitMQ] ，下载rabbitmq\_delayed\_message\_exchange插件，然后解压放置到RabbitMQ的插件目录，进入RabbitMQ的安装目录下的sbin目录，执行下面命令让该插件生效，然后重启RabbitMQ。

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

配置文件：

@Configuration
    public class DelayedRabbitMQConfig {
        public static final String DELAYED_QUEUE_NAME = "delay.queue.demo.delay.queue";
        public static final String DELAYED_EXCHANGE_NAME = "delay.queue.demo.delay.exchange";
        public static final String DELAYED_ROUTING_KEY = "delay.queue.demo.delay.routingkey";
    
        @Bean
        public Queue immediateQueue() {
            return new Queue(DELAYED_QUEUE_NAME);
        }
    
        @Bean
        public CustomExchange customExchange() {
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct");
            return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, args);
        }
    
        @Bean
        public Binding bindingNotify(@Qualifier("immediateQueue") Queue queue,
                                     @Qualifier("customExchange") CustomExchange customExchange) {
            return BindingBuilder.bind(queue).to(customExchange).with(DELAYED_ROUTING_KEY).noargs();
        }
    }

controller层再添加一个入口：

@RequestMapping("delayMsg2")
    public void delayMsg2(String msg, Integer delayTime) {
        log.info("当前时间：{},收到请求，msg:{},delayTime:{}", new Date(), msg, delayTime);
        sender.sendDelayMsg(msg, delayTime);
    }

消息生产者的代码：

public void sendDelayMsg(String msg, Integer delayTime) {
        rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, msg, a ->{
            a.getMessageProperties().setDelay(delayTime);
            return a;
        });
    }

消费者：

@RabbitListener(queues = DELAYED_QUEUE_NAME)
    public void receiveD(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("当前时间：{},延时队列收到消息：{}", new Date().toString(), msg);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

分别按顺序访问：

http://localhost:8080/rabbitmqTest/delayMsg2?msg=msg2&delayTime=20000
    http://localhost:8080/rabbitmqTest/delayMsg2?msg=msg2&delayTime=2000

![20200815221906499.png][]

总结：

延时队列在需要延时处理的场景下非常有用，使用RabbitMQ来实现延时队列可以很好的利用RabbitMQ的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过RabbitMQ集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。延时队列还有很多其它选择，比如利用Java的DelayQueu，利用Redis的zset，利用Quartz或者利用kafka的时间轮。

# 八、如何保证消息的可靠性？ #

一个消息会经历四个节点，只有保证这四个节点的可靠性才能保证整个系统的可靠性。

*  ① 生产者发出后保证到达了MQ。
 *  ② MQ收到消息保证分发到了消息对应的Exchange。
 *  ③ Exchange分发消息入队之后保证消息的持久性。
 *  ④ 消费者收到消息之后保证消息的正确消费。

经历了这四个保证，我们才能保证消息的可靠性，从而保证消息不会丢失。

## 8.1 生产者如何确保消息正确地发送至RabbitMQ？生产者发送消息到MQ失败怎么办？ ##

我们的生产者发送消息之后可能由于网络闪断等各种原因导致我们的消息并没有发送到MQ之中，但是这个时候我们生产端又不知道我们的消息没有发出去，这就会造成消息的丢失。

为了解决这个问题，RabbitMQ引入了**事务机制**和**发送方确认机制（publisher confirm）**，由于事务机制过于耗费性能所以一般不用，这里我着重讲述**发送方确认机制**。

**发送方确认机制（publisher confirm），**就是**消息发送到MQ那端之后，MQ会回一个确认收到的消息**。

打开此功能需要配置：

spring:
      rabbitmq:
        addresses: 127.0.0.1
        host: 5672
        username: guest
        password: guest
        virtual-host: /
        # 打开消息确认机制
        publisher-confirm-type: correlated

生产者：

public void sendAndConfirm() {
            User user = new User();
    
            log.info("Message content : " + user);
    
            CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
            rabbitTemplate.convertAndSend(Producer.QUEUE_NAME,user,correlationData);
            log.info("消息发送完毕。");
    
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    log.info("CorrelationData content : " + correlationData);
                    log.info("Ack status : " + ack);
                    log.info("Cause content : " + cause);
                    if(ack){
                        log.info("消息成功发送，订单入库，更改订单状态");
                    }else{
                        log.info("消息发送失败："+correlationData+", 出现异常："+cause);
                    }
                }
            });
        }

生产者代码里我们看到又多了一个参数：**`CorrelationData`**，这个参数是用来做消息的唯一标识，可以看到是UUID生成的。

打开消息确认之后需要对 **`rabbitTemplate `**多设置一个 `setConfirmCallback `，参数是一个匿名类，消息确认成功 or 失败之后的处理就是写在这个匿名类里面。

比如一条订单消息，当消息确认到达MQ确认之后再行入库或者修改订单的节点状态，如果消息没有成功到达MQ可以进行一次记录或者将订单状态修改。

**ip**：不只有消息没发过去会触发，消息发过去但是找不到对应的Exchange，也会触发。

**即：**

*  生产者把信道设置为 confirm 确认模式，所有在该信道发布的消息都会被指定一个唯一的ID。
 *  一旦消息被投递到目的队列后，或消息被写入磁盘后（可持久化的消息），信道会发送一个确认（Basic.Ack)给生产者（包含消息唯一ID），这样生产者就知道消息到达对应的目的地了。
 *  如果RabbitMQ发生内部错误从而导致消息丢失，会发送一条nack（not acknowledged，未确认）消息。
 *  发送方确认模式是异步的，生产者应用程序在等待确认的同时，可以继续发送消息。当确认消息（Basic.Ack)到达生产者应用程序，生产者应用程序的回调方法就会被触发来处理确认消息。

## 8.2 MQ接收失败或者路由失败怎么办？ ##

生产者的发送消息处理好了之后，MQ端的处理可能出现两个问题：

1.  消息找不到对应的Exchange。
2.  找到了Exchange但是找不到对应的Queue。

这两种情况都可以用** `RabbitMQ `**提供的 **`mandatory `**参数来解决，它会设置消息投递失败的策略，有两种策略：自动删除（**false**）或 返回到客户端(**true）**。

要可靠性，就要 **`mandatory=true`****`，即`** 设置为返回到客户端：

spring:
      rabbitmq:
        addresses: 127.0.0.1
        host: 5672
        username: guest
        password: guest
        virtual-host: /
        # 打开消息确认机制
        publisher-confirm-type: correlated
        # 打开消息返回
        publisher-returns: true
        template:
          mandatory: true

生产者：

public void sendAndReturn() {
            User user = new User();
    
            log.info("Message content : " + user);
    
            rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
                log.info("被退回的消息为：{}", message);
                log.info("replyCode：{}", replyCode);
                log.info("replyText：{}", replyText);
                log.info("exchange：{}", exchange);
                log.info("routingKey：{}", routingKey);
            });
    
            rabbitTemplate.convertAndSend("fail",user);
            log.info("消息发送完毕。");
        }

##  8.3 消息入队之后MQ宕机 ##

MQ突然宕机了或者被关闭了，这种问题就必须要对消息做持久化，以便MQ重新启动之后消息还能重新恢复过来。

消息的持久化要做，但是不能只做消息的持久化，还要做队列的持久化和Exchange的持久化。

@Bean
        public DirectExchange directExchange() {
            // 三个构造参数：name durable autoDelete
            return new DirectExchange("directExchange", false, false);
        }
    
        @Bean
        public Queue erduo() {
            // 其三个参数：durable exclusive autoDelete
            // 一般只设置一下持久化即可
            return new Queue("erduo",true);
        }

创建Exchange和队列时只要设置好持久化，发送的消息默认就是持久化消息。

设置持久化时一定要将Exchange和队列都设置上持久化。

单单只设置Exchange持久化，重启之后队列会丢失。单单只设置队列的持久化，重启之后Exchange会消失，既而消息也丢失，所以如果不两个一块设置持久化将毫无意义。

**Tip：** 这些都是MQ宕机引起的问题，如果出现服务器宕机或者磁盘损坏则上面的手段统统无效，必须引入镜像队列，做异地多活来抵御这种不可抗因素。

## 8.4  如何确保消费者消费了消息？消费者无法正常消费怎么办？ ##

**接收方消息确认机制**：消费者接收每一条消息后都必须进行确认（消息接收和消息确认是两个不同操作）。只有消费者确认了消息，RabbitMQ才能安全地把消息从队列中删除。这里并没有用到超时机制，RabbitMQ仅通过Consumer的连接中断来确认是否需要重新发送消息。也就是说，只要连接不中断，RabbitMQ给了Consumer足够长的时间来处理消息。

spring:
      rabbitmq:
        addresses: 127.0.0.1
        host: 5672
        username: guest
        password: guest
        virtual-host: /
        # 手动确认消息
        listener:
          simple:
              acknowledge-mode: manual

打开手动消息确认之后，只要我们这条消息没有成功消费，无论中间是出现消费者宕机还是代码异常，只要连接断开之后这条信息还没有被消费那么这条消息就会被重新放入队列再次被消费。

下面罗列几种特殊情况：

*  如果消费者接收到消息，在确认之前断开了连接或取消订阅，RabbitMQ会认为消息没有被分发，然后重新分发给下一个订阅的消费者。（可能存在消息重复消费的隐患，需要根据bizId去重）
 *  如果消费者接收到消息却没有确认消息，连接也未断开，则RabbitMQ认为该消费者繁忙，将不会给该消费者分发更多消息。

## 8.5 如何避免消息重复投递或重复消费？ ##

在消息生产时，MQ内部针对每条生产者发送的消息生成一个inner-msg-id，作为去重和幂等的依据（消息投递失败并重传），避免重复的消息进入队列；在消息消费时，要求消息体中必须要有一个bizId（对于同一业务全局唯一，如支付ID、订单ID、帖子ID等）作为去重和幂等的依据，避免同一条消息被重复消费。

## 8.6  如何解决丢数据的问题? ##

当然这一部分会跟上面的内容有些重复，但是是从数据入手的分析过程。

**1.生产者丢数据**

从生产者弄丢数据这个角度来看，RabbitMQ提供transaction和confirm模式来确保生产者不丢消息。

transaction机制：发送消息前，开启事物(channel.txSelect())，然后发送消息，如果发送过程中出现什么异常，事物就会回滚(channel.txRollback())，如果发送成功则提交事物(channel.txCommit())。然而缺点就是吞吐量下降了。

confirm模式：一旦channel进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，rabbitMQ就会发送一个Ack给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了.如果rabiitMQ没能处理该消息，则会发送一个Nack消息给你，生产者可进行重试操作。

**2.消息队列丢数据**

开启持久化磁盘的配置可以防止队列丢失数据。这个持久化配置可以和confirm机制配合使用，可以在消息持久化磁盘后，再给生产者发送一个Ack信号。这样，如果消息持久化磁盘之前，rabbitMQ阵亡了，那么生产者收不到Ack信号，生产者会自动重发。

那么如何持久化呢，就下面两步

①、将queue的持久化标识durable设置为true,则代表是一个持久的队列

②、发送消息的时候将deliveryMode=2

这样设置以后，rabbitMQ就算挂了，重启后也能恢复数据。在消息还没有持久化到硬盘时，可能服务已经死掉，这种情况可以通过引入mirrored-queue即镜像队列，但也不能保证消息百分百不丢失（整个集群都挂掉）

**3.消费者丢数据**

启用手动确认模式可以解决这个问题

①自动确认模式，消费者挂掉，待ack的消息回归到队列中。消费者抛出异常，消息会不断的被重发，直到处理成功。不会丢失消息，即便服务挂掉，没有处理完成的消息会重回队列，但是异常会让消息不断重试。

②手动确认模式，如果消费者来不及处理就死掉时，没有响应ack时会重复发送一条信息给其他消费者；如果监听程序处理异常了，且未对异常进行捕获，会一直重复接收消息，然后一直抛异常；如果对异常进行了捕获，但是没有在finally里ack，也会一直重复发送消息(重试机制)。

③不确认模式，acknowledge="none" 不使用确认机制，只要消息发送完成会立即在队列移除，无论客户端异常还是断开，只要发送完就移除，不会重发。

消费者获取消息的方式：

*  推
 *  拉

## 8.7 消息可靠性方案 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70 4][]

这个例子中的消息是先入库的，然后生产者从DB里面拿到数据包装成消息发给MQ，经过消费者消费之后对DB数据的状态进行更改，然后重新入库。

这中间有任何步骤失败，数据的状态都是没有更新的，这时通过一个定时任务不停的去刷库，找到有问题的数据将它重新扔到生产者那里进行重新投递。

基础的可靠性保证之后，定时任务做一个兜底进行不断的扫描，力图100%可靠性。

# 九、其他 #

## 9.1  vhost 是什么？起什么作用？ ##

每一个RabbitMQ服务器都能创建虚拟的消息服务器，也叫虚拟主机(virtual host)，简称vhost。

默认为“/”。

vhost 可以理解为虚拟 broker ，即 mini-RabbitMQ  server。其内部均含有独立的 queue、exchange 和 binding 等，但最最重要的是，其拥有独立的权限系统，可以做到 vhost 范围的用户控制。当然，从 RabbitMQ 的全局角度，vhost 可以作为不同权限隔离的手段（一个典型的例子就是不同的应用可以跑在不同的 vhost 中）。

## 9.2  RabbitMQ特点? ##

*  **可靠性**: RabbitMQ使用一些机制来保证可靠性， 如持久化、传输确认及发布确认等。
 *  **灵活的路由** : 在消息进入队列之前，通过交换器来路由消息。对典型的路由功能RabbitMQ 己经提供了一些内置的交换器来实现。针对更复杂的路由功能，可以将多个 交换器绑定在一起， 也可以通过插件机制来实现自己的交换器。
 *  **扩展性**: 多个RabbitMQ节点可以组成一个集群，也可以根据实际业务情况动态地扩展 集群中节点。
 *  **高可用性** : 队列可以在集群中的机器上设置镜像，使得在部分节点出现问题的情况下队 列仍然可用。
 *  **多种协议**: RabbitMQ除了原生支持AMQP协议，还支持STOMP， MQTT等多种消息中间件协议。
 *  **多语言客户端** :RabbitMQ 几乎支持所有常用语言，比如 Java、 Python、 Ruby、 PHP、 C\#、 JavaScript 等。
 *  **管理界面** : RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息、集 群中的节点等。
 *  **插件机制**: RabbitMQ 提供了许多插件，以实现从多方面进行扩展，也可以编写自己的插件。

## 9.3  AMQP是什么? AMQP协议3层？AMQP模型的几大组件？ ##

**AMQP是什么?**

RabbitMQ就是 AMQP 协议的 Erlang 的实现(当然 RabbitMQ 还支持 STOMP2、 MQTT3 等协议 ) AMQP 的模型架构 和 RabbitMQ 的模型架构是一样的，生产者将消息发送给交换器，交换器和队列绑定 。

RabbitMQ 中的交换器、交换器类型、队列、绑定、路由键等都是遵循的 AMQP 协议中相应的概念。目前 RabbitMQ 最新版本默认支持的是 AMQP 0-9-1。

**AMQP协议3层：**

Module Layer:协议最高层，主要定义了一些客户端调用的命令，客户端可以用这些命令实现自己的业务逻辑。

Session Layer:中间层，主要负责客户端命令发送给服务器，再将服务端应答返回客户端，提供可靠性同步机制和错误处理。

TransportLayer:最底层，主要传输二进制数据流，提供帧的处理、信道服用、错误检测和数据表示等。

**AMQP模型的几大组件**：

*  交换器 (Exchange)：消息代理服务器中用于把消息路由到队列的组件。
 *  队列 (Queue)：用来存储消息的数据结构，位于硬盘或内存中。
 *  绑定 (Binding)： 一套规则，告知交换器消息应该将消息投递给哪个队列。

## 9.4  交换器无法根据自身类型和路由键找到符合条件队列时，有哪些处理？ ##

mandatory ：true 返回消息给生产者。

mandatory: false 直接丢弃。

## 9.5 优先级队列？ ##

优先级高的队列会先被消费。

可以通过x-max-priority参数来实现。

当消费速度大于生产速度且Broker没有堆积的情况下，优先级显得没有意义。

## 9.6 RabbitMQ 的事务机制？ ##

RabbitMQ 客户端中与事务机制相关的方法有三个:

1.  channel.txSelect 用于将当前的信道设置成事务模式。
2.  channel . txCommit 用于提交事务 。
3.  channel . txRollback 用于事务回滚，如果在事务提交执行之前由于 RabbitMQ 异常崩溃或者其他原因抛出异常，通过txRollback来回滚。

由于事务机制过于耗费性能所以一般不用。

## 9.7 集群中的节点类型？ ##

内存节点：ram,将变更写入内存。

磁盘节点：disc,磁盘写入操作。

RabbitMQ要求最少有一个磁盘节点。

## 9.8 RabbitMQ 中消息可能有的几种状态? ##

alpha: 消息内容(包括消息体、属性和 headers) 和消息索引都存储在内存中 。

beta: 消息内容保存在磁盘中，消息索引保存在内存中。

gamma: 消息内容保存在磁盘中，消息索引在磁盘和内存中都有 。

delta: 消息内容和索引都在磁盘中 。

## 9.9 RabbitMQ 队列结构？ ##

通常由以下两部分组成：

**rabbit\_amqqueue\_process** :负责协议相关的消息处理，即接收生产者发布的消息、向消费者交付消息、处理消息的确认(包括生产端的 confirm 和消费端的 ack) 等。

**backing\_queue**:是消息存储的具体形式和引擎，并向 rabbit amqqueue process 提供相关的接口以供调用。

## 9.10 [消息积压了该如何处理？][Link 1] ##

消息积压的直接原因一定是系统中的某个部分出现了性能问题，来不及处理上游发送的消息，才会导致消息积压。

### [1、优化性能来避免消息积压][Link 1] ###

消费端性能优化

使用消息队列的时候，大部分的性能问题都出现在消费端，如果消费的速度跟不上发送生产消息的速度，就会造成消息积压。如果这种性能倒挂的问题只是暂时的，只要消费端的性能恢复之后，超过发送端的性能，那积压的消息是可以逐渐被消化掉的。

要是消费速度一直比生产速度慢，时间长了，整个系统就会出现问题，要么，消息队列的存储被填满无法提供服务，要么消息丢失，这对于整个系统来说都是严重故障。

在设计系统的时候，一定要保证消费端的消费性能要高于生产端的发送性能。

消费端的性能优化除了优化消费业务逻辑之外，也可以通过水平扩容，增加消费端的并发数来提升总体的消费性能。在扩容Consumer的实例数量的同时，必须同步扩容主题中的分区数量，确保Consumer的实例数和分区数量是相等的。如果Consumer的实例数量超过分区数量，这样的扩容是无效的。

### [2、消息积压了该如何处理？][Link 1] ###

还有一种消息积压的情况是，日常系统正常运转的时候，没有积压或者只有少量积压很快就消费掉了，但是某一时刻，突然就开始积压消息并且积压持续上涨。这种情况下需要在短时间内找到消息积压的原因，迅速解决问题。

能导致积压突然增加，最粗粒度的原因，只有两种：要么是发送变快了，要么是消费变慢了。

大部分消息队列都内置了监控的功能，只要通过监控数据，很容易确定是哪种原因。如果是单位事件发送的消息增多，比如说是赶上大促或者抢购，短时间内不太可能优化消费端的代码来提升消费性能，唯一的方法是通过扩容消费端的实例来提升总体的消费能力。

如果短时间内没有足够的服务器资源进行扩容，没办法的办法是将系统降级，通过关闭一些不重要的业务，减少发送方发送的数据量，最低限度让系统还能正常运转，服务一些重要业务。

还有一种不太常见的情况，通过监控发现，无论是发送消息的速度还是消费消息的速度和原来都没什么变化，这时候需要检查一下消费端是不是消费失败导致的一条消息发福消费这种情况比较多，这种情况也会拖垮整个系统的消费速度。

**消息生产比消费速度快得多，有什么解决方案吗？**

如果要提高消费速度，可以增加多个消费者。

# 补充一：RocketMQ 之 延迟消息原理 #

在延迟队列中，业务队列里的消息到期后，就会被业务队列抛弃，由死信交换机路由到对应的死信队列。

那么业务队列里有很多设置了TTL的队列，业务队列是怎么感知到哪个消息要过期了呢？

在 JDK 中， **`ScheduledThreadPoolExecutor`** 有提供延时消息的功能，但是因为内部的队列使用的是 **`DelayedWorkQueue`** 队列，而 **`DelayedWorkQueue`** 使用的是 **堆**，堆的插入时间复杂度均为 **`O(logn)`**, 这个效率是非常低的，绝大多数的 **MQ** 都无法忍受这样的效率。

因此，MQ 都会重新实现延迟功能。

## 一、**`时间轮 算法`** ##

大多数 MQ比如 Kafka 会使用的 **`时间轮`** 算法（时间轮的应用并非 Kafka 独有，其应用场景还有很多，在 Netty、Akka、Quartz、Zookeeper 等组件中都存在时间轮的踪影。）。

Kafka中的时间轮（TimingWheel）是一个存储定时任务的环形队列，底层采用数组实现，数组中的每个元素可以存放一个定时任务列表（TimerTaskList）。TimerTaskList是一个环形的双向链表，链表中的每一项表示的都是定时任务项（TimerTaskEntry），其中封装了真正的定时任务TimerTask。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5rKZ5rupZGXmtYHmspk_size_18_color_FFFFFF_t_70_g_se_x_16][]

**什么使用环形队列？**

假设我们现在有一个很大的数组，专门用于存放延时任务。它的精度达到了毫秒级！那么我们的延迟任务实际上需要将定时的那个时间简单转换为毫秒即可，然后将定时任务存入其中：

比如说当前的时间是2018/10/24 19:43:45，那么就将任务存入Task\[1540381425000\]，value则是定时任务的内容。

假如这个数组长度达到了亿级，我们确实可以这么干。 那如果将精度缩减到秒级呢？我们也需要一个百亿级长度的数组。

先不说内存够不够，显然你的定时器要这么大的内存显然很浪费。

抛开其他疑问，我们看看手腕上的手表（如果没有去找个钟表，或者想象一个），是不是无论当前是什么时间，总能用我们的表盘去表示它（忽略精度）。

![20210928202226828.png][]

就拿秒表来说，它总是落在 0 - 59 秒，每走一圈，又会重新开始。这样，我们的时间总能落在0 - 59任意一个bucket上，就如同我们的秒钟总是落在0 - 59刻度上一样，这便是**时间轮的环形队列**。

会发现这么一个问题：如果只能表示60秒内的定时任务应该怎么存储与取出，那是不是太有局限性了？**如果想要加入一小时后的延迟任务，该怎么办？**

其实还是可以看一看钟表，对于只有三个指针的表（一般的表）来说，最大能表示12个小时，超过了12小时这个范围，时间就会产生歧义。如果我们加多几个指针呢？比如说我们有秒针，分针，时针，上下午针，天针，月针，年针...... 那不就能表示很长很长的一段时间了？而且，它并不需要占用很大的内存。

比如说秒针我们可以用一个长度为60的数组来表示，分针也同样可以用一个长度为60的数组来表示，时针可以用一个长度为24的数组来表示。那么表示一天内的所有时间，只需要三个数组即可。

> 针对上图的几个名词简单解释下：
> 
>  *  **tickMs**： 时间轮由多个时间格组成，每个时间格就是 tickMs，它代表当前时间轮的基本时间跨度。
>  *  **wheelSize**： 代表每一层时间轮的格数
>  *  **interval**： 当前时间轮的总体时间跨度，interval=tickMs × wheelSize
>  *  **startMs**： 构造当层时间轮时候的当前时间，第一层的时间轮的 startMs 是 TimeUnit.NANOSECONDS.toMillis(nanoseconds()),上层时间轮的 startMs 为下层时间轮的 currentTime。
>  *  **currentTime**： 表示时间轮当前所处的时间，currentTime 是 tickMs 的整数倍（通过 currentTime=startMs - (startMs % tickMs 来保正 currentTime 一定是 tickMs 的整数倍），这个运算类比钟表中分钟里 65 秒分钟指针指向的还是 1 分钟）。currentTime 可以将整个时间轮划分为到期部分和未到期部分，currentTime 当前指向的时间格也属于到期部分，表示刚好到期，需要处理此时间格所对应的 TimerTaskList 的所有任务。

tickMs表示一个刻度，比如说上面说的一秒。wheelSize表示一圈有多少个刻度，即上面说的60。interval表示一圈能表示多少时间，即 tickMs \* wheelSize = 60秒。overflowWheel表示上一层的时间轮，比如说，对于秒钟来说，overflowWheel就表示分钟，以此类推。

### 时间轮中的任务存放 ###

若时间轮的 tickMs=1ms，wheelSize=20，那么可以计算得出 interval 为 20ms。

初始情况下表盘指针 currentTime 指向时间格 0，此时有一个定时为 2ms 的任务插入进来会存放到时间格为 2 的 TimerTaskList 中。随着时间的不断推移，指针 currentTime 不断向前推进，过了 2ms 之后，当到达时间格 2 时，就需要将时间格 2 所对应的 TimeTaskList 中的任务做相应的到期操作。

此时若又有一个定时为 8ms 的任务插入进来，则会存放到时间格 10 中，currentTime 再过 8ms 后会指向时间格 10。

如果同时有一个定时为 19ms 的任务插入进来怎么办？新来的 TimerTaskEntry 会复用原来的 TimerTaskList，所以它会插入到原本已经到期的时间格 1 中。

总之，整个时间轮的总体跨度是不变的，随着指针 currentTime 的不断推进，当前时间轮所能处理的时间段也在不断后移，总体时间范围在 currentTime 和 currentTime+interval 之间。

### 时间轮的升降级 ###

如果此时有个定时为 350ms 的任务该如何处理？直接扩充 wheelSize 的大小么？Kafka 中不乏几万甚至几十万毫秒的定时任务，这个 wheelSize 的扩充没有底线，就算将所有的定时任务的到期时间都设定一个上限，比如 100 万毫秒，那么这个 wheelSize 为 100 万毫秒的时间轮不仅占用很大的内存空间，而且效率也会拉低。Kafka 为此引入了层级时间轮的概念，当任务的到期时间超过了当前时间轮所表示的时间范围时，就会尝试添加到上层时间轮中。

![6eb832013dd1f9e22025cda55cfc2e67.png][]

第一层的时间轮 tickMs=1ms, wheelSize=20, interval=20ms。第二层的时间轮的 tickMs 为第一层时间轮的 interval，即为 20ms。每一层时间轮的 wheelSize 是固定的，都是 20，那么第二层的时间轮的总体时间跨度 interval 为 400ms。以此类推，这个 400ms 也是第三层的 tickMs 的大小，第三层的时间轮的总体时间跨度为 8000ms。

刚才提到的 350ms 的任务，不会插入到第一层时间轮，会插入到 interval=20\*20 的第二层时间轮中，具体插入到时间轮的哪个 bucket 呢？先用 350/tickMs(20)=virtualId(17)，然后 virtualId(17) %wheelSize (20) = 17，所以 350 会放在第 17 个 bucket。

如果此时有一个 450ms 后执行的任务，那么会放在第三层时间轮中，按照刚才的计算公式，会放在第 0 个 bucket。第 0 个 bucket 里会包含\[400,800)ms 的任务。随着时间流逝，当时间过去了 400ms，那么 450ms 后就要执行的任务还剩下 50ms 的时间才能执行，此时有一个时间轮降级的操作，将 50ms 任务重新提交到层级时间轮中，那么此时 50ms 的任务根据公式会放入第二个时间轮的第 2 个 bucket 中，此 bucket 的时间范围为\[40,60)ms，然后再经过 40ms，这个 50ms 的任务又会被监控到，此时距离任务执行还有 10ms，同样将 10ms 的任务提交到层级时间轮，此时会加入到第一层时间轮的第 10 个 bucket，所以再经过 10ms 后，此任务到期，最终执行。

kafka 的延迟队列使用时间轮实现，能够支持大量任务的高效触发，但是在 kafka 延迟队列实现方案里还是看到了 delayQueue 的影子，使用 delayQueue 是对时间轮里面的 bucket 放入延迟队列，以此来推动时间轮滚动，但是基于将插入和删除操作则放入时间轮中，将这些操作的时间复杂度都降为 O(1)，提升效率。Kafka 对性能的极致追求让它把最合适的组件放在最适合的位置。

**二、`Timer` 实现**

`RocketMQ` 在延迟消息的实现上更是取巧，没有使用大多数 MQ 会使用的 **`时间轮`** 算法，而是简单的通过 `Timer` 实现。

`Broker` 如何感知到消息已经到期，可以让 `Consumer` 消费或者丢入死信队列，这是重点。下面，来看看 `RocketMQ` 的取巧处理。

`Broker` 端可以通过配置 `messageDelayLevel` 来改变 `RocketMQ` 默认延迟等级配置。下面就是 `RocketMQ` 默认的配置

messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

也就是说，`RocketMQ` 其实无法像其他 `MQ` 一样提供灵活的延迟消息，必须通过配置文件配置。

延迟消息的实现，均在该类下 `ScheduleMessageService。`

`Broker` 启动时，会解析 `messageDelayLevel` 配置，并将值放入 `delayLevelTable` map 中。

public class ScheduleMessageService extends ConfigManager {
    
        public boolean parseDelayLevel() {
    
            HashMap<String, Long> timeUnitTable = new HashMap<String, Long>();
    
            timeUnitTable.put("s", 1000L);
    
            timeUnitTable.put("m", 1000L * 60);
    
            timeUnitTable.put("h", 1000L * 60 * 60);
    
            timeUnitTable.put("d", 1000L * 60 * 60 * 24);
    
            String levelString = this.defaultMessageStore.getMessageStoreConfig().getMessageDelayLevel();
    
            try {
    
                String[] levelArray = levlString.split(" ");
    
                for (int i = 0; i < levelArray.length; i++) {
    
                    String value = levelArray[i];
    
                    String ch = value.substring(value.length() - 1);
    
                    Long tu = timeUnitTable.get(ch);
    
                    int level = i + 1;
    
                    if (level > this.maxDelayLevel) {
    
                        this.maxDelayLevel = level;
    
                    }
    
                    long num = Long.parseLong(value.substring(0, value.length() - 1));
    
                    long delayTimeMillis = tu * num;
    
                    this.delayLevelTable.put(level, delayTimeMillis);
    
                }
    
            } catch (Exception e) {
    
                log.error("parseDelayLevel exception", e);
    
                log.info("levelString String = {}", levelString);
    
                return false;
            }
            return true;
        }
    }

Timer 启动

public void start() {
        if (started.compareAndSet(false, true)) {
            this.timer = new Timer("ScheduleMessageTimerThread", true);
            // 获取配置的延迟等级
            for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
                Integer level = entry.getKey();
                Long timeDelay = entry.getValue();
                // 每个延迟等级的消费偏移
                Long offset = this.offsetTable.get(level);
                if (null == offset) {
                    offset = 0L;
                }
    
                // 初始化延迟调度任务
    
                if (timeDelay != null) {
                    this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
                }
            }
        }
    }

`offsetTable` 是在 `Broker` 启动的时候，从 `${storePath}/store/config/delayOffset.json` 解析出来的。该文件存储的内容是一个 `json`, 存放着每个延迟等级对应的 的消费偏移。大致如下：

{
    "offsetTable":{1:10}
    }

`Timer` 中的任务是 `DeliverDelayedMessageTimerTask`, 接下来看下该类的 `run()` 方法实现逻辑。

`run()` 执行的逻辑如下

1.  根据 延迟队列的 消费偏移，从对应队列中获取消息
2.  根据 `ConsumeQueue` 子条目中的 `tagsCode` 拿到消息存储时的时间戳
3.  将 `tagsCode` 与当前时间对比，如果小于等于当前时间，则将延迟消息恢复为原消息，供 `Consumer` 消费
4.  继续调度下一个延迟消息

# 补充二：RocketMQ集群如何平滑扩缩容？ #

集群的部署如下下图所示：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5rKZ5rupZGXmtYHmspk_size_20_color_FFFFFF_t_70_g_se_x_16][]

![9f23e6076788dce4ebfbb7bf8e4aed6c.png][]

a、扩容Broker和Consumer，都是需要将信息注册到nameServer中。而Producer只有获取到最新的Broker信息时才会根据算法（随机、轮询、hash等）写数据。所以扩容对生产者无影响。  
b、缩容Broker时，只需要里面的数据全部消费完后在去掉即可，缩容的可能性不大。

[6ef4a1a08153ea7f2b032853770e1bb2.png]: /images/20230528/6f64a536c0f44ffc85e8846744db8314.png
[--_Queue_520_-CSDN]: https://blog.csdn.net/weixin_41231928/article/details/104840335
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70]: /images/20230528/79acbd6cc71c450abaf9ac8eb6cfac24.png
[format_png]: /images/20230528/d82e8318e0e84d22a6a64ba2f06bb052.png
[format_png 1]: /images/20230528/399dbe0badc84937bcf17a7870d4f03c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70 1]: /images/20230528/de0eab8ac44e4ac0913c07a5cd32e862.png
[RabbitMQ]: https://www.sojson.com/tag_rabbitmq.html
[format_png 2]: /images/20230528/b76293ad8f5e4c98a69e08e2faa2f92b.png
[format_png 3]: /images/20230528/4fba21d79f3a47f2847e182838675328.png
[d2d51d71282ba3536180a0d2b47f3f6a.png]: /images/20230528/8364fff673164cc282b49dc37fffb863.png
[a3982540411b4afeb7a65088dc09ea3f.png]: /images/20230528/d9f8c099917a42c79c83a8968f740bc8.png
[980910fc84bcc0de248d3f2f45b94e8a.png]: /images/20230528/7f9d496b6d764bd1b873ac3f1f5f86f0.png
[20f7c46a497ccda6d84cb65cb2440bed.png]: /images/20230528/10aacb63e37c437e86349e92ed4e1328.png
[e53c3a3c296337ee5a88bbc974fc8ef3.png]: /images/20230528/3d029f9a7bce4166b5bc8cfa51039d40.png
[761b452420b966d9681750bf72ffc12b.png]: /images/20230528/fec21db15ddc49e1bfd254491194a8b5.png
[300c65e8d2f0c96ef4e0ce0ed2db1959.png]: /images/20230528/8623cff8295c478d8ceff6f3b26207c7.png
[20200317093054938.png]: /images/20230528/6bffffcb376d4129b1a2388cc306ef2e.png
[20200815211455810.png]: /images/20230528/f19aadd3dc084ff991883767a1ab35da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70 2]: /images/20230528/fde5848fecd14667a9dcbdcfa08f30b5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70 3]: /images/20230528/f4bb5abcf82c4026b0af51bbaf4577ad.png
[Community Plugins _ RabbitMQ]: https://www.rabbitmq.com/community-plugins.html
[20200815221906499.png]: /images/20230528/bf7e45e9cd2144abb3d0bacfdcc81c2e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIzMTkyOA_size_16_color_FFFFFF_t_70 4]: /images/20230528/91b898e6eb8e42d986929bc860f51860.png
[Link 1]: https://mp.weixin.qq.com/s?__biz=MzUzMTA2NTU2Ng==&mid=2247487551&idx=1&sn=18f64ba49f3f0f9d8be9d1fdef8857d9&scene=21#wechat_redirect
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5rKZ5rupZGXmtYHmspk_size_18_color_FFFFFF_t_70_g_se_x_16]: /images/20230528/3593cff9f2814a54b1c64d2d9b17048d.png
[20210928202226828.png]: /images/20230528/9ecce4bfccf94a7db9066994cfbb7568.png
[6eb832013dd1f9e22025cda55cfc2e67.png]: /images/20230528/d77ccc5fe5b64d86b100c578f49efccc.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5rKZ5rupZGXmtYHmspk_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/20211002125034488.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5rKZ5rupZGXmtYHmspk=,size_20,color_FFFFFF,t_70,g_se,x_16
[9f23e6076788dce4ebfbb7bf8e4aed6c.png]: https://img-blog.csdnimg.cn/img_convert/9f23e6076788dce4ebfbb7bf8e4aed6c.png