8-Spring Boot消息服务
在实际项目开发中，有时候需要与其他系统进行集成完成相关业务功能，这种情况最原始的做法是程序内部相互调用，除此之外，还可以使用消息服务中间件进行业务处理，使用消息服务中间件处理业务能够提升系统的异步通信和扩展解耦能力。Spring Boot对消息服务管理提供了非常好的支持。
一、消息服务概述
1.为什么使用消息服务
在多数应用尤其是分布式系统中，消息服务是不可或缺的重要部分，它使用起来比较简单，同时解决了不少难题，例如异步处理、应用解耦、流量削锋、分布式事务管理等，使用消息服务可以实现一个高性能、高可用、高扩展的系统。
a.异步处理
b.应用解耦
c.流量削峰
服务器处理资源能力有限，出现峰值时很容易造成服务器宕机、用户无法访问的情况。通常采用消息队列缓冲瞬时高峰流量，对请求进行分层过滤。
d.分布式事务管理
分布式事务管理流程：
2.常用的消息中间件
二、RabbitMQ 消息中间件
1.RabbitMQ 简介
RabbitMQ是基于AMQP协议的轻量级、可靠、可伸缩和可移植的消息代理，Spring使用RabbitMQ通过AMQP协议进行通信，在Spirng Boot中对RabbitMQ进行了集成管理。
2.消息代理过程
3.RabbitMQ 工作模式介绍
a.Work queues（工作队列模式）
b.Publish/Subscribe（发布订阅模式）
c.Routing（路由模式）
d.Topics（通配符模式）
e.RPC
f.Headers（使用较少，不进行详细介绍）
三、安装及使用RabbitMQ
1.下载
在RabbitMQ官网上http://www.rabbitmq.com/install-windows.html
下载,如果是在Windows环境下安装RabbitMQ消息中间件还需要64位的Erlang语言包支持。
2.windows下安装
RabbitMQ安装包依赖于Erlang语言包的支持，所以要先安装Erlang语言包，再安装RabbitMQ安装包。
3.使用
RabbitMQ可视化效果展示, RabbitMQ默认提供了两个端口号5672和15672，其中5672用作服务端口号，15672用作可视化管理端口号。
在浏览器上访问http://localhost:15672(用户名和密码均为guest)
4.可能会出现如下问题：
a.重新安装erlang的otp_win64_21.2.exe时， epmd.exe 装不上？
从任务管理器结束掉epmd.exe任务。
b.RabbitMQ启动后无法访问http://localhost:15672?
确保已切换到RabbitMQ安装目录的sbin目录下，再使用下面操作命令
# 开启RabbitMQ节点
rabbitmqctl start_app
# 开启RabbitMQ管理模块的插件，并配置到RabbitMQ节点上
rabbitmq-plugins enable rabbitmq_management
# 关闭RabbitMQ节点
rabbitmqctl stop
四、Spring Boot整合RabbitMQ环境搭建
1.创建Spring Boot项目，添加Web依赖以及RabbitMQ依赖。
2、在application.properties配置文件中编写需要设置的配置属性。
# 配置RabbitMQ消息中间件连接配置
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
#配置RabbitMQ虚拟主机路径/，默认可以省略
spring.rabbitmq.virtual-host=/
五、Publish/Subscribe（发布订阅模式）
1.基于API的方式及测试的搭建步骤
a.使用AmqpAdmin定制消息组件。
1、在测试类中先引入AmqpAdmin管理类,定制Publish/Subscribe工作模式所需的消息组件
@Autowired
private AmqpAdmin amqpadmin;
/**
* 使用AmqpAdmin管理员API定制消息组件
*/
@Test
public void amqpAdmin() {
// 1、定义fanout类型的交换器
amqpAdmin.declareExchange(new FanoutExchange("fanout_exchange"));
// 2、定义两个默认持久化队列，分别处理email和sms
amqpAdmin.declareQueue(new Queue("fanout_queue_email"));
amqpAdmin.declareQueue(new Queue("fanout_queue_sms"));
// 3、将队列分别与交换器进行绑定
amqpAdmin.declareBinding(new Binding("fanout_queue_email",Binding.DestinationType.QUEUE,"fanout_exchange","",null));
amqpAdmin.declareBinding(new Binding("fanout_queue_sms",Binding.DestinationType.QUEUE,"fanout_exchange","",null));
}
2、执行上述单元测试方法amqpAdmin()，验证RabbitMQ消息组件的定制效果，通过RabbitMQ可视化管理页面的Exchanges面板查看效果。
3、在RabbitMQ可视化管理页面的Exchanges面板中新出现了一个名称为fanout_exchange的交换器,且其类型是设置的fanout类型。单击fanout_exchange交换器进入查看。
4、切换到Queues面板页面，查看定制生成的消息队列信息
b.消息发送者发送消息
1、创建一个实体类User,将User对象作为消息的内容
public class User {
private Integer id;
private String username;
//自己补充其他
2、在项目测试类中使用Spring框架提供的RabbitTemplate模板类实现消息发送。
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 1、Publish/Subscribe工作模式消息发送端
*/
@Test
public void psubPublisher() {
User user=new User();
user.setId(1);
user.setUsername("石头");
rabbitTemplate.convertAndSend("fanout_exchange","",user);
}
3、执行上述消息发送的测试方法psubPublisher()，将出现异常，控制执行效果如图
这个转换器只支持Spring、byte[]和Serializable序列化后的消息
异常的两种解决方式：
- 将实体类实现JDK自带的Serializable序列化接口，不推荐，RabbitMQ管理页面显示消息不够有好
- 定制其他类型的消息转换器
创建一个RabbitMQ消息配置类RabbitMQConfig,并在该配置类中通过@Bean注解自定义了一个Jackson2JsonMessageConverter类型的消息转换器组件，该组件的返回值必须为MessageConverter类型
@Configuration
public class RabbitMQConfig {
/**
* 定制JSON格式的消息转换器
* @return
*/
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
执行上述测试方法，执行成功，查看RabbitMQ可视化管理页面Queues面板信息，两个消息队列中各自拥有一条待接收的消息。
单击某个队列的详情页面，查看具体的信息
c.消息消费者接收消息
消息消费者接受消息,创建一个针对RabbitMQ消息中间件进行消息接收和处理的业务类RabbitMQService，在该类中编写如下方法
@Service
public class RabbitMQService {
/**
* Publish/Subscribe工作模式接收，处理邮件业务
* @param message
*/
@RabbitListener(queues = "fanout_queue_email")
public void psubConsumerEmail(Message message) {
byte[] body = message.getBody();
String s = new String(body);
System.out.println("邮件业务接收到消息： "+s);
}
/**
* Publish/Subscribe工作模式接收，处理短信业务
* @param message
*/
@RabbitListener(queues = "fanout_queue_sms")
public void psubConsumerSms(Message message) {
byte[] body = message.getBody();
String s = new String(body);
System.out.println("短信业务接收到消息： "+s);
}
启动项目来监听并接收消息队列中的消息。程序启动成功后，立即查看控制台打印结果
2.基于配置类的方式
a.打开RabbitMQ消息配置类RabbitMQConfig，定义fanout类型的交换器、不同名称的消息队列以及将不同名称的消息队列与交换器绑定。
/**
* 使用基于配置类的方式定制消息中间件
* @return
*/
// 1、定义fanout类型的交换器
@Bean
public Exchange fanout_exchange(){
return ExchangeBuilder.fanoutExchange("fanout_exchange").build();
}
// 2、定义两个不同名称的消息队列
@Bean
public Queue fanout_queue_email(){
return new Queue("fanout_queue_email");
}
@Bean
public Queue fanout_queue_sms(){
return new Queue("fanout_queue_sms");
}
// 3、将两个不同名称的消息队列与交换器进行绑定
@Bean
public Binding bindingEmail(){
return BindingBuilder.bind(fanout_queue_email()).to(fanout_exchange()).with("").noargs();
}
@Bean
public Binding bindingSms(){
return BindingBuilder.bind(fanout_queue_sms()).to(fanout_exchange()).with("").noargs();
}
2.重新运行测试类的psubPublisher方法，来发送信息，之前定义的消息消费者会消费这个消息。
3.基于注解的方式
该方式定义了消息组件、消息消费者，只要发送消息，就可以测试了。
1、打开业务类RabbitMQService，使用@RabbitListener注解及其相关属性定制消息发送组件,在该类中编写处理邮件以及短信任务的方法
/**
* **使用基于注解的方式实现消息服务
* 1.1、Publish/Subscribe工作模式接收，处理邮件业务
* @param user
*/
@RabbitListener(bindings =@QueueBinding(value =@Queue("fanout_queue_email"), exchange =@Exchange(value = "fanout_exchange",type = "fanout")))
public void psubConsumerEmailAno(User user) {
System.out.println("邮件业务接收到消息： "+user);
}
/**
* 1.2、Publish/Subscribe工作模式接收，处理短信业务
* @param user
*/
@RabbitListener(bindings =@QueueBinding(value =@Queue("fanout_queue_sms"),exchange =@Exchange(value = "fanout_exchange",type = "fanout")))
public void psubConsumerSmsAno(User user) {
System.out.println("短信业务接收到消息： "+user);
}
2.重新运行测试类的psubPublisher方法，来发送信息。
六、Routing(路由模式)
1.使用基于注解的方式定制消息组件和消息消费者
打开业务类RabbitMQService，在该类中使用@RabbitListener注解及其相关属性定制Routing路由模式的消息组件，并模拟编写消息消费者接收的方法。
/**
* 2.1、路由模式消息接收，处理error级别日志信息
* @param message
*/
@RabbitListener(bindings =@QueueBinding(value =@Queue("routing_queue_error"),exchange =@Exchange(value = "routing_exchange",type = "direct"),key = "error_routing_key"))
public void routingConsumerError(String message) {
System.out.println("接收到error级别日志消息： "+message);
}
/**
* 2.2、路由模式消息接收，处理info、error、warning级别日志信息
* @param message
*/
@RabbitListener(bindings =@QueueBinding(value =@Queue("routing_queue_all"),exchange =@Exchange(value = "routing_exchange",type = "direct"),key = {"error_routing_key","info_routing_key","warning_routing_key"}))
public void routingConsumerAll(String message) {
System.out.println("接收到info、error、warning等级别日志消息： "+message);
}
2.消息发送者发送消息
在测试类中使用RabbitTemplate模板类实现Routing路由模式下的消息发送
/**
* 2、Routing工作模式消息发送端
*/
@Test
public void routingPublisher() {
rabbitTemplate.convertAndSend("routing_exchange","error_routing_key","routing send error message");
}
直接执行上述消息发送的测试方法routingPublisher()，控制台效果如图
将测试方法routingPublisher()中进行消息发送的参数进行修改，调整发送info级别的日志信息（注意同时修改info_routing_key路由键），
再次启动该测试方法，查看控制台执行效果，如图
通过RabbitMQ可视化管理页面查看自动定制的Routing路由模式的消息组件，如图所示
七、Topics(通配符模式)
1.使用基于注解的方式定制消息组件和消息消费者
打开业务类RabbitMQService，在该类中使用@RabbitListener注解及其相关属性定制Topics通配符模式的消息组件，并模拟编写消息消费者接收的方法。
/**
* 3.1、通配符模式消息接收，进行邮件业务订阅处理
* @param message
*/
@RabbitListener(bindings =@QueueBinding(value =@Queue("topic_queue_email"),exchange =@Exchange(value = "topic_exchange",type = "topic"),key = "info.#.email.#"))
public void topicConsumerEmail(String message) {
System.out.println("接收到邮件订阅需求处理消息： "+message);
}
/**
* 3.2、通配符模式消息接收，进行短信业务订阅处理
* @param message
*/
@RabbitListener(bindings =@QueueBinding(value =@Queue("topic_queue_sms"),exchange =@Exchange(value = "topic_exchange",type = "topic"),key = "info.#.sms.#"))
public void topicConsumerSms(String message) {
System.out.println("接收到短信订阅需求处理消息： "+message);
}
2.消息发送者发送消息
在项目测试类Chapter08ApplicationTests中使用RabbitTemplate模板类实现Routing路由模式下的消息发送。
/**
* 3、Topcis工作模式消息发送端
*/
@Test
public void topicPublisher() {
// 1、只发送邮件订阅用户消息
// rabbitTemplate.convertAndSend("topic_exchange","info.email","topics send email message");
// 2、只发送短信订阅用户消息
// rabbitTemplate.convertAndSend("topic_exchange","info.sms","topics send sms message");
// 3、发送同时订阅邮件和短信的用户消息
rabbitTemplate.convertAndSend("topic_exchange","info.email.sms","topics send email and sms message");
}
执行测试方法topicPublisher()，先进行邮件订阅用户的消息发送
进行短信订阅用户的消息发送
同时进行邮件和短信订阅用户的消息发送方法
通过RabbitMQ可视化管理页面查看自动定制的Topics通配符模式的消息组件，使用基于注解的方式自动生成了Topics通配符模式下的消息组件，并进行了自动绑定。
8-Spring Boot消息服务相关推荐
- Spring Boot 消息队列 RocketMQ 入门
转载自 芋道 Spring Boot 消息队列 RocketMQ 入门 摘要: 原创出处 http://www.iocoder.cn/Spring-Boot/RocketMQ/ 「芋道源码」欢迎转载 ...
- Spring Boot微服务的黑匣子测试是如此简单
当我需要进行原型设计,概念验证或在空闲时间使用一些新技术时,开始新项目对于Maven来说总是有点烦人. 不得不说,设置Maven项目并不难,您可以使用Maven原型. 但是原型通常是过时的. 谁想玩旧 ...
- Spring Boot 微服务编码风格指南和最佳实践
文奇摄于世界尽头州立公园 通过多年来使用 Spring Boot 微服务,我编制了一份编码风格指南和最佳实践列表.这份清单并不全面,但我希望您能找到一两点可以借鉴的地方,无论您是新手还是经验丰富的 S ...
- spring boot 微服务集群 + 注册中心
spring boot 微服务框架下载地址: https://start.spring.io/ 注册中心 Eureka Server提供服务注册服务,各个节点启动后,会在Eureka Server中进 ...
- Docker容器及Spring Boot微服务应用
2019独角兽企业重金招聘Python工程师标准>>> Docker容器及Spring Boot微服务应用 1 什么是Docker 1.1 Docker的出现 问题一:项目实施环境复 ...
- Spring Boot 消息
Spring Boot 消息 转载于:https://www.cnblogs.com/Aha-Best/p/11205641.html
- 高级版的 jvisualvm ：Spring Boot Admin 监控 Spring Boot 微服务项目
前奏:先说一下 Java VisualVM Java VisualVM 是一个能够监控 JVM 的 jdk 自带的图形化工具: 在 $JAVA_HOME/bin 目录下,可直接运行它. 要想监控远程服 ...
- Docker基础篇 - （六）Docker 网络Spring Boot微服务打包Docker镜像
⑦ Docker 网络 7.1 理解Docker0 清空下前面的docker 镜像.容器 # 删除全部容器 [root@cVzhanshi tomcat-diy]# docker rm -f $(do ...
- Spring Boot微服务间文件返回实现
Feign接口获取文件流问题_Java_wyazyf的博客-CSDN博客 https://blog.csdn.net/wyazyf/article/details/93200033 Spring Bo ...
最新文章
- 不用等到2050年，现在已经有人跟机器人结婚了
- python分组求和_Python学习笔记之pandas索引列、过滤、分组、求和功能示例
- Windows下Visual studio 2013 编译 Audacity
- Flask基础(03)--创建第一个Flask程序
- Jquery 实现原理之 Ajax
- jpa 多层嵌套一对多_JPA一对多、多对多json序列化死循环问题解决
- 前端架构 IMVC 实时热更新模式
- Codeforces Round #453 (Div. 1)
- 应用时间序列分析--基于Eviews软件
- Mysql实战练习之简单图书管理系统
- 8. 求 s=a+aa+aaa+aaaa+aa…a 的值
- 2021年计算机软考时间公布啦
- 【免费扩容onedrive】
- 人工智能-数据分析-鸢尾花
- 设计模式-设计原则-开闭原则是目标，里氏代换原则是基础，依赖倒转原则是手段...
- 心拔凉拔凉五连追问No.58
- 光学识别OCR软件Readiris Pro 17 for Mac
- 不知道自己不知道 知道自己不知道 不知道自己知道 知道自己知道
- 学习突围1 - 关于学习的痛点
- Ajax基础：发送jsonp请求（jsonp、jsonCallback）
热门文章
- 深度学习入门笔记：Day-10
- CentOS 7 安装 Python 3.6
- 【算法】克鲁斯卡尔算法
- 【kafka】kafka 如何 删除 清除 消费组 信息
- 95-38-040-Buffer-AbstractReferenceCountedByteBuf引用计数
- 【es】es 分布式一致性原理剖析(二)-Meta篇
- 10-30-010-安全简介-Kafka 安全机制
- 【Siddhi】Siddhi 4.2.40版本之前的一个Bug
- 【Flink】The class WC is an instance class, meaning it is not a member of a toplevel object
- 01-windows下 Rserve安装