当前位置：首页 > news >正文

RabbitMQ 2025/3/5

news 来源：原创 2025/8/29 10:48:10

高性能异步通信组件。

同步调用

以支付为例：

可见容易发生雪崩。

异步调用

以支付为例：

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

消息代理（英文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（消息直接写磁盘里）。它不仅仅具有数据持久化的优势，同时还解决了并发能力下降的问题。