云南专业网站优化,佛山广告设计公司排名,安阳市商祺网络有限责任公司,宿迁房产文章目录 1. 补充知识#xff1a;同步通讯和异步通讯1.1 同步通讯1.2 异步通讯 2. 同步调用的缺点2.1 业务耦合2.2 性能较差2.3 级联失败 3. 什么情况下使用同步调用4. 异步调用5. 异步调用的优点和缺点5.1 异步调用的优点5.1.1 解除耦合#xff0c;拓展性强5.1.2 无需等待同步通讯和异步通讯1.1 同步通讯1.2 异步通讯 2. 同步调用的缺点2.1 业务耦合2.2 性能较差2.3 级联失败 3. 什么情况下使用同步调用4. 异步调用5. 异步调用的优点和缺点5.1 异步调用的优点5.1.1 解除耦合拓展性强5.1.2 无需等待性能好5.1.3 故障隔离5.1.4 削峰填谷 5.2 异步调用的缺点5.2.1 不能得到调用结果5.2.2 不确定下游业务执行是否成功5.2.3 业务安全依赖于消息代理的可靠性 6. MQ 的技术选型7. 安装 RabbitMQ 并启动 RabbitMQ7.1 搜索 RabbitMQ 镜像7.2 下载 RabbitMQ 镜像7.3 启动 RabbitMQ7.4 访问 RabbitMQ 的管理页面7.5 可能遇到的问题7.5.1 安全组和防火墙未开放端口7.5.2 RabbitMQ 没有安装 Web 插件 8. RabbitMQ 的整体架构和核心概念9. RabbitMQ 快速入门9.1 新建队列9.2 绑定队列与交换机9.3 发送消息9.4 可能遇到的问题 10. 数据隔离10.1 新建用户10.2 为新用户创建一个 VirtualHost10.3 测试不同 VirtualHost 之间是否有数据隔离 11. 在 SpringBoot 项目中集成 RabbitMQ11.1 AMQP 和 SpringAMQP11.2 快速入门11.2.1 引入 Maven 依赖11.2.2 编写与 RabbitMQ 有关的配置信息 11.3 完成一个简单的案例11.3.1 创建队列11.3.2 发送消息11.3.3 接收消息 12. Work Queues 模型12.1 Work Queues 的概念12.2 Work Queues 模型的消息推送机制 13. 交换机13.1 Fanout 交换机13.1.1 Fanout 交换机的概念13.1.2 快速上手 13.2 Direct 交换机13.2.1 Direct 交换机的概念13.2.2 快速上手 13.3 Topic 交换机推荐使用13.3.1 Topic 交换机的概念13.3.2 快速上手 14. 在 SpringBoot 项目中声明队列和交换机的方式14.1 编程式声明14.1.1 SpringAQMP提供的创建队列和交换机的类14.1.2 快速上手14.1.3 编程式声明的缺点 14.2 注解式声明推荐使用 15. 消息转换器15.1 默认的消息转换器15.2 自定义消息转换器 1. 补充知识同步通讯和异步通讯 1.1 同步通讯
同步通讯是指发送方在发送消息后会等待接收方的回应直到收到回应后才会继续执行后续操作 同步通讯的特点是
阻塞发送方在等待回应期间会被阻塞无法进行其他操作顺序执行消息的处理是按照发送和接收的顺序进行的确保了消息的时序性实时反馈发送方可以立即得到接收方的回应适用于需要立即确认的场景占用资源由于需要等待可能会造成资源的浪费如线程阻塞
打电话就是一个典型的同步通讯例子通话双方必须实时交流一方说话时另一方必须等待
1.2 异步通讯
异步通讯是指发送方在发送消息后不需要等待接收方的立即回应就可以继续执行其他操作。接收方在处理完消息后可能会在未来的某个时间点给出回应 异步通讯的特点是
非阻塞发送方在发送消息后可以立即继续其他工作不会因为等待回应而被阻塞解耦发送方和接收方在时间上解耦可以独立处理各自的任务灵活异步通讯可以处理更复杂的通信模式如消息队列、事件驱动等资源利用率高更高效地利用资源因为不需要等待可以提高系统的吞吐量
电子邮件是一个异步通讯的例子你可以发送一封邮件后继续做其他事情收件人可以在任何时间回复邮件微信聊天也是一个异步通讯的例子
2. 同步调用的缺点
我们以支付业务为例分析同步调用的缺点
支付业务采用的是同步调用的方式因为我们在执行更新支付状态操作和更新订单状态之前需要先知道扣减余额操作的结果这种同步调用方式存在几个问题
2.1 业务耦合
第一个问题就是业务耦合的问题对于支付服务来说最重要的一件事就是扣减用户的余额然后更新支付状态
后续的更新订单状态操作跟支付服务是没什么关系的但是支付成功之后确实需要更新订单状态所以支付服务不得不调用交易服务来更新订单状态
那有同学就说了我在支付服务里面加一行代码不就可以调用交易服务了吗听起来没什么问题但是业务是会变化的产品经理的脑洞你也是想象不到的
想象一下产品经理提了一个新的需求用户支付成功之后要发一个短信通知用户产品经理一提需求我们就要更改源代码 某一天产品经理又提了一个新需求用户支付成功之后要为用户增加一定的积分 这种同步调用的方式拓展性比较差不符合面向对象编程中的开闭原则
2.2 性能较差
如果采用同步调用的方式支付服务需要等待其它所有服务完成操作耗时会大大增长十分影响用户的体验 2.3 级联失败
想象一下交易服务出现故障了而这个故障迟迟没有得到解决最终就很有可能拖垮支付服务导致支付服务的资源被耗尽也出现故障出现级联失败的情况 3. 什么情况下使用同步调用
经过上面的分析有同学可能会有这样的疑问既然同步调用有这么多问题为什么我们还要用同步调用呢什么情况下使用同步调用呢 一般来说使用同步调用的场景都有一个特点下一步操作依赖于上一步操作的结果 以上面的支付业务为例交易服务、通知服务、积分服务都依赖于支付服务的结果
当支付服务成功扣减用户余额并成功更新支付状态之后交易服务、通知服务、积分服务就可以开始执行相应的操作了
然而通知服务不依赖于交易服务积分服务也不依赖于通知服务
在成功扣减用户余额并成功更新支付状态之后支付业务就已经完成了
所以说支付服务完成了之后只需要通知交易服务、通知服务、积分服务执行相应的操作而不需要等待交易服务、通知服务、积分服务都完成之后再返回结果
4. 异步调用
异步调用基于消息通知一般包含三个角色消息
发送者投递消息的人消息代理管理、暂存、转发消息的人消息接收者接收和处理消息的人 改为异步调用之后支付服务不再同步调用与支付业务关联度低的服务而是发送消息通知于支付业务关联度低的服务 5. 异步调用的优点和缺点
5.1 异步调用的优点
5.1.1 解除耦合拓展性强
即使以后有新业务拓充支付服务只需要发送一条消息给消息代理让消息代理通知新业务拓展性强
5.1.2 无需等待性能好
支付服务完成之后只需要发送消息给消息代理让消息代理通知其它服务
5.1.3 故障隔离
即使交易服务出现了故障也不会影响到支付服务 5.1.4 削峰填谷
假如支付服务正在面临着很大的压力流量时高时低呈波浪形。如果采用同步调用的方式当流量很高的时候交易服务、通知服务、积分服务可能扛不住
但如果采用异步调用的方式就很少会出现交易服务、通知服务、积分服务扛不住的情况。为什么呢因为消息代理容量很大。在高并发的情况下用户每成功支付一次支付服务只需要发送一条消息给消息代理这些像洪水一般的消息都会被消息代理拦住
消息代理会保存这些消息后续服务可以根据自己的处理速度从消息代理中一条一条地取出信息并处理。这样一来后续服务承受的压力将会变得很平缓 5.2 异步调用的缺点
5.2.1 不能得到调用结果
异步调用一般是通知对方执行某个操作无法知道对方执行操作后的结果
5.2.2 不确定下游业务执行是否成功
异步调用通知下游业务后无法知道下游业务是否执行成功
5.2.3 业务安全依赖于消息代理的可靠性
下游业务的执行依赖于消息代理的可靠性一旦消息代理出现故障下游业务将无法执行
6. MQ 的技术选型
MQMessage Queue消息队列
以下是当前主流的消息队列 在这里重点提一下 Kafka Kafka 的吞吐量非常高适合大规模日志场景 目前国内大部分公司采用的都是 RabbitMQ
7. 安装 RabbitMQ 并启动 RabbitMQ
RabbitMQ是基于 Erlang 语言开发的开源消息通信中间件官网RabbitMQ
我们基于 docker 安装 RabbitMQ
7.1 搜索 RabbitMQ 镜像
sudo docker search rabbitmq7.2 下载 RabbitMQ 镜像
sudo docker pull rabbitmq检查 RabbitMQ 镜像是否下载成功
sudo docker images7.3 启动 RabbitMQ
sudo docker run \-e RABBITMQ_DEFAULT_USERwuyanzu \-e RABBITMQ_DEFAULT_PASSbhoLdSvpd0UAOysh \-v rabbitmq-plugins:/plugins \--name rabbitmq \--hostname rabbitmq \-p 15672:15672 \-p 5672:5672 \-d \rabbitmq:latest指令说明 sudo docker run: 基本的Docker命令用于启动一个新的容器实例 -e RABBITMQ_DEFAULT_USERwuyanzu: 设置RabbitMQ服务的默认用户名为wuyanzu -e RABBITMQ_DEFAULT_PASSkZoeSW$$xS5i^Cum: 设置RabbitMQ服务的默认密码为bhoLdSvpd0UAOysh -v rabbitmq-plugins:/plugins: 将一个名为rabbitmq-plugins的卷映射到容器的/plugins目录用于存放RabbitMQ的插件。这里的rabbitmq-plugins是一个卷的名称而不是宿主机的路径 --name rabbitmq: 指定容器的名称为rabbitmq --hostname rabbitmq: 设置容器的主机名为rabbitmq -p 15672:15672: 将宿主机的端口15672映射到容器的端口15672这是RabbitMQ管理界面的默认端口 -p 5672:5672: 将宿主机的端口5672映射到容器的端口5672这是RabbitMQ用于AMQP协议通信的默认端口 -d: 在后台运行容器守护进程 rabbitmq:latest: 使用最新的RabbitMQ官方镜像来创建容器
7.4 访问 RabbitMQ 的管理页面
接下来进入 RabbitMQ 的管理界面在浏览器输入以下地址将 IP 地址换成你的虚拟机的 IP 地址
http://127.0.0.1:15672/输入用户名和密码后进入到 RabbitMQ 的管理页面 7.5 可能遇到的问题
7.5.1 安全组和防火墙未开放端口
如果无法进入RabbitMQ的管理界面记得先在安全组和防火墙中开放 15672 和 5672 端口
在 Ubuntu 中开放15672 和 5672 端口
sudo ufw allow 15672
sudo ufw allow 5672
sudo ufw reload在 CentOS 中开放15672 和 5672 端口
sudo firewall-cmd --zonepublic --add-port15672/tcp --permanent
sudo firewall-cmd --zonepublic --add-port5672/tcp --permanent
sudo firewall-cmd --reload7.5.2 RabbitMQ 没有安装 Web 插件
如果开放防火墙端口后还是无法访问 RabbitMQ 的管理界面可能是安装 RabbitMQ 没有安装 Web 插件
以下是 RabbitMQ 安装 Web 插件的方法
第一步进入容器内部
sudo docker exec -it rabbitmq bash第二步安装 Web 插件
rabbitmq-plugins enable rabbitmq_management安装插件后退出容器内部
exit8. RabbitMQ 的整体架构和核心概念
RabbitMQ 有几个核心概念
Publisher消息发送者Consumer消息的消费者Queue消息队列存储消息Exchange交换机负责路由消息VirtualHost虚拟主机用于数据隔离 RabbitMQ 的整体架构 9. RabbitMQ 快速入门 注意事项交换机只能路由和转发消息不能存储消息 9.1 新建队列
创建一个名为 hello.queue 的队列 9.2 绑定队列与交换机
我们将刚才新创建的 hello.queue 队列与 amq.fanout 交换机绑定fanout意为扇出 绑定成功后的界面 9.3 发送消息
我们在 amq.fanout 交换机中发送一条消息消息的内容为 Hello, RabbitMQ! 发送消息后查看交换机的总览信息 查看队列中的消息数 查看消息的具体内容 9.4 可能遇到的问题
如果你发现
交换机的 overview 页面没有折线图Queues 页面也没有与消息相关的信息点击channels后出现Stats in management UI are disabled on this node信息
需要先修改 RabbitMQ的 配置 第一步进入容器内部
sudo docker exec -it rabbitmq bash第二步修改配置
cd /etc/rabbitmq/conf.d/echo management_agent.disable_metrics_collector false management_agent.disable_metrics_collector.conf第三步重启容器
先退出容器内部
exit再重启容器
sudo docker restart rabbitmq10. 数据隔离
10.1 新建用户
新建一个名为 CaiXuKun 的用户密码为 T1rhFXMGXIOYCoyi 角色指定为 admin 可以看到新用户对任意一个 VirtualHost 都是没有访问权限的 用新用户的账号登录管理台虽然能看到所有 VirtualHost 的信息但是无法对任意一个 VirtualHost 进行操作 10.2 为新用户创建一个 VirtualHost
用新用户的账号登录管理台创建一个名为 /blog 的 VirtualHost 10.3 测试不同 VirtualHost 之间是否有数据隔离
可以看到不同的 VirtualHost 之间有不同的交换机 对某一个 VirtualHost 操作不会影响到另一个 VirtualHost
11. 在 SpringBoot 项目中集成 RabbitMQ
后端环境
SpringBoot3.0.2JDK17.0.7
11.1 AMQP 和 SpringAMQP SpringAMQP 的官网Spring AMQP
11.2 快速入门
新建一个 SpringBoot 项目并创建 consumer 和 publisher 两个子模块项目的整体结构如下 11.2.1 引入 Maven 依赖
在父工程中引入 SpringAMQP 的依赖
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency11.2.2 编写与 RabbitMQ 有关的配置信息
在 consumer 和 publisher 模块的 application.yml 中分别编写与 RabbitMQ 有关的配置信息
spring:rabbitmq:host: 127.0.0.1port: 5672virtual-host: /blogusername: CaiXuKunpassword: T1rhFXMGXIOYCoyi11.3 完成一个简单的案例
案例要求如下
在 RabbitMQ 的控制台中创建名为 simple.queue 的队列队列归属的 VirtualHost 为 /blog在 publisher 模块中利用 SpringAMQP 直接向 simple.queue 队列发送消息在 consumer 服务中利用 SpringAMQP 编写消费者监听 simple.queue 队列
11.3.1 创建队列 11.3.2 发送消息
在 publisher 模块中编写测试类用户向队列发送消息
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;SpringBootTest(classes PublisherApplication.class)
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testvoid testSendMessageToQueue() {String queueName simple.queue;String msg Hello, SpringAMQP!;rabbitTemplate.convertAndSend(queueName, msg);}}在 RabbitMQ 的控制台可以看到消息成功发送 11.3.3 接收消息
SpringAMQP 提供了声明式的消息监听我们只需要通过RabbitListener注解在方法上声明要监听的队列名称将来 SpringAMQP 就会把消息传递给使用了RabbitListener注解的方法
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class RabbitMQListener {RabbitListener(queues simple.queue)public void listenSimpleQueue(String message) {System.out.println(消费者收到了 simple.queue 的消息【 message 】);}}启动 consumer 模块的启动类之后就可以看到消息的内容
12. Work Queues 模型
12.1 Work Queues 的概念
Work Queues简单地来说就是让多个消费者绑定到一个队列共同消费队列中的消息
虽然有多个消费者绑定同一个队列但是队列中的某一条消息只会被一个消费者消费 我们实现一个小案例模拟 Work Queues实现一个队列绑定多个消费者
案例要求如下
在RabbitMQ的控制台创建一个队列名为 work.queue在 publisher 服务中定义测试方法在 1 秒内产生 50 条消息发送到work.queue在 consumer 服务中定义两个消息监听者都监听 work.queue 队列消费者 1 每秒处理 50 条消息消费者 2 每秒处理 5 条消息
在 publisher服务的 SpringAmqp 测试类中添加以下方法该方法可以在 1 秒内产生 50 条消息
Test
void testWorkQueue() throws InterruptedException {String queueName work.queue;for (int i 1; i 50; i) {String message Hello, work queues_ i;rabbitTemplate.convertAndSend(queueName, message);Thread.sleep(20);}
}在 consumer 服务的 RabbitMQListener 类中添加以下方法监听 work.queue 队列
RabbitListener(queues work.queue)
public void listenWorkQueue1(String message) {System.out.println(消费者1 收到了 work.queue的消息【 message 】);
}RabbitListener(queues work.queue)
public void listenWorkQueue2(String message) {System.err.println(消费者2 收到了 work.queue的消息...... 【 message 】);
}12.2 Work Queues 模型的消息推送机制
如果有两个或两个以上的消费者监听同一个队列默认情况下 RabbitMQ 会采用轮询的方法将消息分配给每个队列
但每个消费者的消费能力可能是不一样的我们给两个消费者中的代码设置不同的休眠时间模拟消费能力的不同
RabbitListener(queues work.queue)
public void listenWorkQueue1(String message) throws InterruptedException {System.out.println(消费者1 收到了 work.queue的消息【 message 】);Thread.sleep(20);
}RabbitListener(queues work.queue)
public void listenWorkQueue2(String message) throws InterruptedException {System.err.println(消费者2 收到了 work.queue的消息...... 【 message 】);Thread.sleep(100);
}经过测试可以发现即使两个队列的消费能力不一样默认情况下 RabbitMQ 还是会采用轮询的方法将消息分配给每个队列也就是平均分配
但这不是我们想要的效果我们想要的效果是消费能力强的消费者处理更多的消息甚至能够帮助消费能力弱的消费者
怎么样才能达到这样的效果呢只需要在配置文件中添加以下信息
spring:rabbitmq:listener:simple:prefetch: 1这个配置信息相当于告诉消费者要一条一条地从队列中取出消息只有处理完一条消息才能取出下一条
这样一来就可以充分利用每一台机器的性能让消费能力强的消费者处理更多的消息同时还可以避免消息在消费能力较弱的消费者上发生堆积的情况
13. 交换机
真正的生产环境都会经过交换机来发送消息而不是直接发送到队列
交换机的作用
接收 publisher 发送的消息将消息按照规则路由到与交换机绑定的队列 交换机的类型有以下三种
Fanout广播Direct定向Topic话题 注意事项交换机只能路由和转发消息不能存储消息 13.1 Fanout 交换机
13.1.1 Fanout 交换机的概念
Fanout 交换机会将接收到的消息广播到每一个跟其绑定的 queue 所以也叫广播模式 13.1.2 快速上手
我们做一个小案例来体验 Fanout 交换机的效果案例要求如下
在 RabbitMQ 控制台中声明队列 fanout.queue1 和 fanout.queue2在 RabbitMQ 控制台中声明交换机 blog.fanout将两个队列与其绑定在 consumer 服务中编写两个消费者方法分别监听 fanout.queue1 和 fanout.queue2 队列在 publisher 服务中编写测试方法向 blog.fanout 交换机发送消息 声明交换机 在 consumer 服务的 RabbitMQListener 类中添加以下方法分别监听 fanout.queue1 和 fanout.queue2 队列
RabbitListener(queues fanout.queue1)
public void listenFanoutQueue1(String message) {System.out.println(消费者1 收到了 fanout.queue1的消息【 message 】);
}RabbitListener(queues fanout.queue2)
public void listenFanoutQueue2(String message) {System.err.println(消费者2 收到了 fanout.queue2的消息...... 【 message 】);
}在 publisher 服务的 SpringAmqp 测试类中添加以下方法向 blog.fanout 交换机发送消息
Test
void testSendFanout() {String exchangeName blog.fanout;String message Hello, fanout exchange;rabbitTemplate.convertAndSend(exchangeName, null, message);
}13.2 Direct 交换机
13.2.1 Direct 交换机的概念
Direct 交换机会将接收到的消息根据规则路由到指定的队列被称为定向路由
每一个 Queue 都与 Exchange 设置一个 bindingKey发布者发送消息时指定消息的 RoutingKeyExchange 将消息路由到 bindingKey 与消息 routingKey 一致的队列 需要注意的是同一个队列可以绑定多个 bindingKey 如果有多个队列绑定了同一个 bindingKey 就可以实现类似于 Fanout 交换机的效果。由此可以看出Direct 交换机的功能比 Fanout 交换机更强大
13.2.2 快速上手
我们做一个小案例来体验 Direct 交换机的效果案例要求如下
在 RabbitMQ 控制台中声明队列 direct.queue1 和 direct.queue2在 RabbitMQ 控制台中声明交换机 blog.direct 将上面创建的两个队列与其绑定在 consumer 服务中编写两个消费者方法分别监听 direct.queue1 和 direct.queue2在 publisher 服务中编写测试方法利用不同的 RoutingKey 向 blog.direct 交换机发送消息 为 direct.queue1队列 和 direct.queue2 队列分别指定 bindingKey 在 consumer 服务的 RabbitMQListener 类中添加以下方法分别监听 direct.queue1 和 direct.queue2 队列
RabbitListener(queues direct.queue1)
public void listenDirectQueue1(String message) {System.out.println(消费者1 收到了 direct.queue1的消息【 message 】);
}RabbitListener(queues direct.queue2)
public void listenDirectQueue2(String message) {System.err.println(消费者2 收到了 direct.queue2的消息...... 【 message 】);
}在 publisher 服务的 SpringAmqp 测试类中添加以下方法向 blog.direct交换机发送消息
Test
void testSendDirect() {String exchangeName blog.direct;String blueMessage 蓝色通知警报解除哥斯拉放的是气球;rabbitTemplate.convertAndSend(exchangeName, blue, blueMessage);String redMessage 红色警报由于日本排放核污水惊现哥斯拉!;rabbitTemplate.convertAndSend(exchangeName, red, redMessage);String yellowMessage 黄色通知哥斯拉来了快跑;rabbitTemplate.convertAndSend(exchangeName, yellow, yellowMessage);
}13.3 Topic 交换机推荐使用
13.3.1 Topic 交换机的概念
Topic Exchange 与 Direct Exchange类似区别在于 Topic Exchange 的 routingKey 可以是多个单词的列表多个 routingKey 之间以.分割
Queue 与 Exchange 指定 bindingKey 时可以使用通配符
#代指 0 个或多个单词*代指 1 个单词 Topic 交换机能实现的功能 Direct 交换机也能实现不过用 Topic 交换机实现起来更加方便如果某条消息的 topic 符合多个 queue 的 bindingKey 该条消息会发送给符合条件的所有 queue 实现类似于 Fanout 交换机的效果 13.3.2 快速上手
我们做一个小案例来体验 Topic 交换机的效果案例要求如下
在 RabbitMQ 控制台中声明队列 topic.queue1 和 topic.queue2在 RabbitMQ 控制台中声明交换机 blog.topic 将两个队列与其绑定在 consumer 服务中编写两个消费者方法分别监听 topic.queue1 和 topic.queue2在 publisher 服务中编写测试方法利用不同的 routingKey 向 blog.topic 发送消息 为 topic.queue1 和 topic.queue2 队列分别指定 bindingKey 在 consumer 服务的 RabbitMQListener 类中添加以下方法分别监听 direct.queue1 和 direct.queue2 队列
RabbitListener(queues topic.queue1)
public void listenTopicQueue1(String message) {System.out.println(消费者1 收到了 topic.queue1的消息【 message 】);
}RabbitListener(queues topic.queue2)
public void listenTopicQueue2(String message) {System.err.println(消费者2 收到了 topic.queue2的消息...... 【 message 】);
}在 publisher 服务的 SpringAmqp 测试类中添加以下方法向 blog.direct交换机发送消息
Test
void testSendTopic() {String exchangeName blog.topic;String weatherMessage 今天天气挺不错我的心情的挺好的;rabbitTemplate.convertAndSend(exchangeName, china.weather, weatherMessage);String newsMessage 蓝色通知警报解除哥斯拉放的是气球;rabbitTemplate.convertAndSend(exchangeName, china.news, newsMessage);
}14. 在 SpringBoot 项目中声明队列和交换机的方式
我们之前创建队列和交换机都是在 RabbitMQ 的控制台页面中创建的不仅十分繁琐还有可能打错队列和交换机的名。而且不同的环境开发环境、测试环境、生产环境可能会有不同的队列和交换机手动创建队列和交换机效率十分低下
接下来为大家介绍两种在 SpringBoot 项目中声明队列和交换机的方式
14.1 编程式声明
14.1.1 SpringAQMP提供的创建队列和交换机的类
SpringAMQP 提供了几个类用来声明队列、交换机及其绑定关系
Queue用于声明队列可以用工厂类 QueueBuilder 构建Exchange用于声明交换机可以用工厂类 ExchangeBuilder 构建Binding用于声明队列和交换机的绑定关系可以用工厂类 BindingBuilder 构建 14.1.2 快速上手
我们创建一个 Fanout 类型的交换机并且创建队列与这个交换机绑定
在 consumer 服务中编写 FanoutConfiguration 配置类
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class FanoutConfiguration {Beanpublic FanoutExchange fanoutExchange3() {return ExchangeBuilder.fanoutExchange(blog.fanout3).build();}Beanpublic FanoutExchange fanoutExchange4() {return new FanoutExchange(blog.fanout4);}Beanpublic Queue fanoutQueue3() {return new Queue(fanout.queue3);}Beanpublic Queue fanoutQueue4() {return QueueBuilder.durable(fanout.queue4).build();}Beanpublic Binding fanoutBinding3(Queue fanoutQueue3, FanoutExchange fanoutExchange3) {return BindingBuilder.bind(fanoutQueue3).to(fanoutExchange3);}Beanpublic Binding fanoutBinding4() {return BindingBuilder.bind(fanoutQueue4()).to(fanoutExchange4());}}启动 consumer 的启动类之后队列、交换机、队列和交换机之间的关系就会自动创建 创建 Queue 时如果没有指定 durable 属性则 durable 属性默认为 true 14.1.3 编程式声明的缺点
编程式声明有一个缺点当队列和交换机之间绑定的 routingKey 有很多个时编码将会变得十分麻烦
以下是一个队列与 Direct 类型交换机绑定两个 routingKey 时的代码
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class DirectConfiguration {Beanpublic DirectExchange directExchange3() {return new DirectExchange(blog.direct3);}Beanpublic Queue directQueue3() {return new Queue(direct.queue3);}Beanpublic Queue directQueue4() {return new Queue(direct.queue4);}Beanpublic Binding directQueue3BindingRed(Queue directQueue3, DirectExchange directExchange3) {return BindingBuilder.bind(directQueue3).to(directExchange3).with(red);}Beanpublic Binding directQueue3BindingBlue(Queue directQueue3, DirectExchange directExchange3) {return BindingBuilder.bind(directQueue3).to(directExchange3).with(blue);}Beanpublic Binding directQueue4BindingRed(Queue directQueue4, DirectExchange directExchange3) {return BindingBuilder.bind(directQueue4).to(directExchange3).with(red);}Beanpublic Binding directQueue4BindingBlue(Queue directQueue4, DirectExchange directExchange3) {return BindingBuilder.bind(directQueue4).to(directExchange3).with(yellow);}}14.2 注解式声明推荐使用
SpringAMOP 提供了基于RabbitListener注解声明队列和交换机的方式 我们先在 RabbitMQ 的控制台删除 blog.direct 交换机、 direct.queue1 队列和 direct.queue2 队列
再改造 consumer 服务的 RabbitMQListener 类的监听 direct.queue1 队列和 direct.queue2 队列的方法
RabbitListener(bindings QueueBinding(value Queue(name direct.queue1),exchange Exchange(name blog.direct, type ExchangeTypes.DIRECT),key {red, blue}
))
public void listenDirectQueue1(String message) {System.out.println(消费者1 收到了 direct.queue1的消息【 message 】);
}RabbitListener(bindings QueueBinding(value Queue(name direct.queue2),exchange Exchange(name blog.direct, type ExchangeTypes.DIRECT),key {red, yellow}
))
public void listenDirectQueue2(String message) {System.out.println(消费者2 收到了 direct.queue2的消息【 message 】);
}15. 消息转换器
在了解消息转换器之前我们先来做一个小案例案例的内容是利用 SpringAMQP 发送一条消息消息的内容为一个 Java 对象
案例要求如下
在 RabbitMQ 控制台创建一个队列名为 object.queue编写单元测试向该队列中直接发送一条消息消息的内容为 Map在控制台查看消息 在 publisher 服务的 SpringAmqpTests 测试类中新增 testSendObject 方法
Test
void testSendObject() {MapString, Object hashMap new HashMap(2);hashMap.put(name, Tom);hashMap.put(age, 21);rabbitTemplate.convertAndSend(object.queue, hashMap);
}成功发送消息后我们在 RabbitMQ 的控制台查看消息的具体内容 可以发现消息的内容类型为 application/x-java-serialized-object并且消息的内容也变成一堆乱码
我们本来是想发送一个简单的仅含有姓名和年龄两个字段的简短信息但是消息却变成了一堆乱码不仅可读性大大下降而且占用的空间也大大地增加了这显然不是我们想要的效果
15.1 默认的消息转换器
Spring 处理对象类型的消息是由 org.springframework.amap.support.converter.MessageConverter 接口来处理的该接口默认实现是 SimpleMessageConverter
SimpleMessageConverter 类是基于 JDK 提供的 ObjectOutputStream 来类完成序列化的这种序列化方式存在以下问题:
使用 JDK 序列化有安全风险如果序列化后的消息被恶意篡改在反序列化的过程中可能会执行一些高危的代码经过 JDK 序列化的消息占用空间很大经过 JDK 序列化的消息可读性很差
15.2 自定义消息转换器
一般建议采用 JSON 序列化代替默认的 JDK 序列化
要使用 JSON 序列化需要先引入 jackson 依赖在项目的父工程中引入
如果是 Web 项目无需引入该依赖因为 spring-boot-starter-web 依赖中已包含该依赖
dependencygroupIdcom.fasterxml.jackson.dataformat/groupIdartifactIdjackson-dataformat-xml/artifactId
/dependency接着在 consumer 服务和 publisher 服务中配置 MessageConverter
Bean
public MessageConverter jacksonMessageConvertor(){return new Jackson2JsonMessageConverter();
}再次发送对象类型的消息可以看到消息已经成功转换成 JSON 类型的字符串 我们也可以在 consumer 服务的 RabbitMQListener 类中添加对 object.queue 队列的监听用什么类型发就用什么类型接收
RabbitListener(queues object.queue)
public void listenObjectQueue(MapString, Object hashMap) {System.out.println(消费者收到了 object.queue的消息【 hashMap 】);
}启动 consumer 服务的启动类之后在控制台中可以看到被转换成 JSON 格式的消息
在控制台中会看到报错信息因为之前有一条用 JDK 序列化的消息现在改用了 jackson 序列化序列化和反序列化用的序列化器不一样肯定会报错
报错后消息就没了出现了消息丢失的现象该怎么解决呢可以参考我的另一篇博文RabbitMQ 高级篇
