RabbitMQ 2025/3/5

高性能异步通信组件。

同步调用

以支付为例：

可见容易发生雪崩。

异步调用

以支付为例：

支付服务当甩手掌柜了，不管后面的几个服务的结果。只管库库发，后面那几个服务想取的时候就取，因为消息代理里可以一直装，缓存消息。

消息代理（英文Broker）

消息代理相关的技术MQ技术

Erlang面向并发的语言

 

RabbitMQ安装部署

可直接采用Docker，方便。

RabbitMQ消息流转的过程（整体架构）

消费者监听队列，发送者不直接发给队列，而是发给exchange交换机，交换机会根据规则把消息路由给不同队列。

因为RabbitMQ的性能很强，每秒钟可以达到数万的并发，所以企业有多个项目的话，往往部署一套RabbitMQ就够了，多个项目可以共享RabbitMQ服务。但是大家一起的话，很可能交换机出现冲突，所以引出RabbitMQ的新概念virtual-host(虚拟主机)，类似与MySQL里的database（众所周知，MySQL里可以创建多个database，每个database的表和其他database的表直接是相互隔离的），每个项目都创一个自己的virtual-host，就可以相互隔离开了。

例子：MQ入门-06.RabbitMQ-快速入门_哔哩哔哩_bilibili

交换机不存只负责转发。交换机和队列必须有一个关系，才能给队列发消息。binding绑定关系。

数据隔离

虚拟主机是实现了数据隔离。

不同的项目创建不同的用户。为新建用户建一个虚拟主机。

视频举例：MQ入门-07.RabbitMQ-数据隔离_哔哩哔哩_bilibili

RabbitMQ的java客户端

这里我们不采用Rabbit官方提供的java客户端，而是Spring AMQP，它是基于AMQP协议（消息收发与语言和平台无关），官方提供的java客户端使用起来繁琐，所以使用Spring AMQP。

4:00  MQ入门-08.Java客户端-快速入门_哔哩哔哩_bilibili

因为是简单入门案例，可以省去交换机这个步骤。

控制台创建队列->pom里引入spring-amqp依赖->yaml里配置RabbitMQ服务端信息（如地址主机名、端口、虚拟主机名、用户、密码），这样微服务才能连接到RabbitMQ->发送消息（SpringAMQP提供了RabbitTemplate工具类，方便我们发送消息）->接受消息

接收者接受消息的代码：加一个@Comonent把它注册成Spring的一个Bean，这个类的内部要有一个方法，这个方法要加上@RabbitListener（也就是消费者的监听者）的注解后面再带上队列名，现在只要队列有消息，方法就可以拿到了。方法参数自己设。

WorkQueues

任务模型，让多个消费者绑定到一个队列，共同消费队列中的消息。

那么队列中的消息会被哪个消费者收到呢？

模拟WorkQueue（实现一个队列绑定多个消费者）

MQ入门-09.Java客户端-WorkQueue_哔哩哔哩_bilibili

在RabbitMQ控制台建立一个队列->一个发消息的->2个消费者

下图两个消费者，一个发送者。

 

队列同一个消息只能被一个消费者处理，很多条消息的话均匀分配（默认轮询，1人1条）。（上图2个消费者结果有奇偶规律）

把很多消息平均分给消费者，可以加快消息处理速度。

每个消费者能力不一样（通过sleep(ms)修改），均匀分配消息（默认）肯定是不合理的，只需要修改application.yml，设置preFetch值为1，实现能者多劳。

Fanout交换机（广播交换机）

再看一下以前学的这个发消息的过程。带有交换机的完整模型。

交换机的作用是接受发送者发送的消息，并将消息路由到与其绑定的队列。

那么什么是Fanout交换器？

特征：把接收到的消息路由到每一个和他绑定的队列。（队列中的消息只能被一个消费者处理，有了Fanout交换机，发的消息就可以被多个消费者处理了。我们完全可以给每个微服务创建一个队列，然后队列绑到交换机上，fanout交换机想广播一样，给每个队列群发/复制消息。）

案例步骤：声明2个队列，一个交换机exhanges，然后banding绑定，第一种方法可以直接用控制台发消息然后查看，还有一种方法java代码，2段接受代码，1段发送代码（这回调3个参数的api，exchange，null，message，之前写的例子用到的都是两个参数，队列名和消息）。

Direct交换机（定向）

MQ入门-11.Java客户端-Direct交换机_哔哩哔哩_bilibili

和原来差不多，这回交换机选direct，队列设置bindingkey，发送时发送者的参数分别设好exchange，routingkey值，message。其余的步骤和原来一样。

所以两个key值一样，direct交换机也能实现Fanout交换机的功能。一个queue可以设多个bindingkey。

Topic交换机（话题）

 队列bindingkey可以通过通配符简易设置。

topic和direct比除了多了一个统配符，功能差不多。

声明队列交换机

之前队列和交换机的创建都是依靠控制台，这次学习用代码声明队列交换机，这样项目一启动就会自己创建队列和交换机了。

可以用new的方式。（更简单）

也可以用builder方式。

发送者只管发，什么也不关心，所以通常我们在消费者这一端，声明队列交换机及绑定关系。

步骤（以Fanout交换机为例）：创建FanoutConfiguration类，声明交换机，队列，绑定关系。可以new（比较简单），也可以builder构建（比较专业）。

上述基于JavaBean绑定太麻烦了，这回学习注解@RatbbitListener。

基于JavaBean还是基于注解，完全个人喜好了。（但是如果基于javabean，声明direct交换机好像没法写很多bindingKey）

消息转换器

以前我们发消息一直用到rabbitTemplate的convertAndSend方法。我们发消息是可以传任何java对象作为消息，网络传输其实是以字节传的，直接对象不行，因此传的时候要把java对象转换成字节。这个转换就是由消息转换器转的。

java里有一种JDK自带的序列化的一个工具（能把任一java对象序列化成字节的形式），所以这个消息转换的过程就是采用JDK自带的序列化方式。

ObjectOutputStream：对象流，jdk自带的序列化工具，能把任意java对象序列化成字节。

但是推荐使用JSON的消息转换器。

发送方和接收方一定要用相同的消息转换器。

pom配置好之后到MAVEN里刷新一下。

给发送方和接收方都配置一个消息转换类。

对比图（下边的JSON的消息转换器）

业务改造

把业务从OpenFeign的同步调用改成基于MQ的异步调用。

OpenFeign：02-基本概念_哔哩哔哩_bilibili

nacos可以抽取共享配置，不用重复进行同样的配置。

共同代码可以写到common里

消息可靠性

发送者的可靠性

发送者重连

yml

发送者确认

路由失败的两种原因：

exchange2交换机没连队列，路由失败。返回ACK（因为消息确实发出去了）

routingkey和bandingkey没有匹配上的。

如果返回NACK要重发消息。

none是关闭（默认）

simple是同步阻塞

correlated异步的

至此，ConfirmCallback和ReturnCallback全部写完了。

唯一，uuid，最简单的随机算法

MQ的可靠性

消息传到MQ也不一样可靠，因为MQ本身也可能把消息弄丢。

MQ内存储存的方式，可能弄丢消息，还有可能导致MQ性能下降和阻塞。

为此，产生两个方案：数据持久化（把数据提前持久化存到磁盘，提前的意思：不是等到满了再存到磁盘，提前就开始往磁盘存了）、LazyQueue

数据持久化

持久化的一种方法：写出到磁盘，这样就永久保存了。

交换机的持久化（默认都是持久的交换机、durable属性）（Spring的AMQP代码生成的交换机默认也是持久化的）

队列持久化（默认持久化durable属性）（Spring的AMQP代码生成的默认也是持久化的）

消息持久化（手动设置为Persistent）

LazyQueue

上节课写了数据持久化，当我们把交换机、队列、消息持久化了以后，就不用再担心MQ宕机而导致消息丢失了，不仅如此，RabbitMQ也不会再因为消息堆积配置out而出现阻塞了。但是数据持久化后，不仅要在内存里写，还要再磁盘里写一份，这样每条消息处理的耗时就增加了，这也就导致它的整体并发能力有点下降。为了解决这个问题，引入了一种新的队列模式，lazy Queue（消息直接写磁盘里）。它不仅仅具有数据持久化的优势，同时还解决了并发能力下降的问题。

那么怎样去设置一个队列变成Lazy Queue模式呢

第一种方法，控制台添加

第二种方法，代码添加（声明bean的方式或者注解都可以）

消费者的可靠性

消费者确认机制

三种状态（ack，nack，reject）：不管哪种状态，都不能在刚收到消息的时候就返回，我们应该根据处理结果去做判断，也就是consumer处理完的时候。

注意：这回配置的是消费者下的application文件。

有个缺点：消息异常了就会像踢球一样反复被踢来踢去，下面学的失败重试机制就可以解决这个问题（设置最大尝试次数）

失败重试机制

第一种：重试次数耗尽后直接reject，不要了

第二种：重试次数耗尽后，返回nack，消息回到队列，再重试

第三种：重试次数耗尽后，消费者将失败消息扔进另一个交换机，这个交换机连着另一个队列

下面以第三种方式为例（相比于前两种更复杂，多了一个交换机和一个新队列）

业务幂等性

解决重复消费问题（比如事务发到消费者那了，本来消费者要给MQ返回一个ACK，因为断电没发送出去，MQ以为消费者宕机了，于是恢复电的时候，MQ又会重新给消费者发送一次，可消费者已经处理完这个事务了，如果这个事务涉及到支付，那消费者就会白白消费两次。）这时就用到业务幂等性了。

下面以消息重复提交导致业务被重复执行的这个场景为例的几个方案：

1.唯一id（我们可以给消息带上唯一的id）

这种方案存在业务侵入的问题：本来是没有id这个属性的，现在的消息都带了id属性，并且还放在了数据库里！！！！！

2.业务判断

加了一个业务判断，当订单来了后，不急着标记订单为已支付，而是每次都在标记前查询一下订单状态。

延迟消息

在上面三个可靠性都失败下的兜底方案——延迟消息

死信交换机

一旦我们通过dead-letter-exchange属性指定了一个交换机，那么死信就都会被投到这个交换机里，不会被丢弃了。

我们已经有了一组交换机和队列，再准备一组特殊的交换机和队列，给上面的队列通过dead-letter-exchange属性指定下面的交换机，现在下面的交换机就变成上面的队列的死信交换机了。我们不会给上面的队列绑定消费者，通常是给死信交换机的队列绑定消费者。发送者向上面的交换机发消息，发消息时给消息设置过期时间TTL=30s，然后消息就发到上面的队列，因为上面的队列没有消费者，就在这里停下来了开始等待，有个计时器开始等待数30秒，30秒后消息变成死信进入死信交换机。这样就通过死信交换机的形式实现了延迟效果。

延迟消息插件

用了这个插件，只要简单的把delay属性配置成true就好了，然后就可以实现延迟消息。

步骤：首先下载配置好插件->声明交换机并设置其delay属性为true(可用注解或bean方式)->发消息的时候要设置一个过期时间

利用延迟消息解决用户订单超时未支付问题

可以分成几步：下单之后发消息，完成延迟消息的发送。编写消息监听，它要干的事是去修改订单状态。利用延迟消息实现延时任务的效果。

MQ高级-13.延迟消息-取消超时订单_哔哩哔哩_bilibili