SpringBoot使用RabbitMQ
*有话在先*
介绍优点缺点安装配置RabbitMQ简单的使用RabbitMQ多对多对于传对象的支持不同类型的ExchangeTopicHeadersFanout
demo参考文档：
有话在先
本文章比较长，就是为了能详细弄清每行代码有什么用
介绍
RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控制、负载均衡等特性，主要是为了实现系统之间的双向解耦
优点
解耦、异步提升效率、流量削峰[高并发]
缺点
系统的可用性降低：引入的外部依赖越多，系统越容易崩掉
系统的复杂性提高：考虑的问题会变的多很多，如何保证消息没有重复消费？如何保证消息不丢失？怎么保证消息传递的顺序？
一致性问题：假设ABC系统，A系统发送完消息直接返回成功，但是BC系统之中若有系统写库失败，则会产生数据不一致
安装
安装并配置erlang
安装地址：https://www.erlang.org/downloads
新建系统变量：ERLANG_HOME=安装路径[能看到bin目录的]
安装RabbitMQ
安装地址：https://www.rabbitmq.com/install-windows.html
新建系统变量：RABBITMQ_SERVER=安装路径/rabbitmq_server-版本
在path中添加：%RABBITMQ_SERVER%\sbin
安装RabbitMQ可视化工具 在RabbitMQ的路径下/sbin目录下执行
安装web可视化工具 rabbitmq-plugins.bat enable rabbitmq_management
安装服务 rabbitmq-service.bat install
启动服务 rabbitmq-server.bat start
访问：localhost:15672 可访问控制台，用户密码为guest
配置RabbitMQ
添加pom 在原有的springboot基础配置的dependency的基础上加上
配置Springboot配置文件 在resources的application.yml中添加 # application.yml
spring:
application:
name: Spring-boot-rabbitmq
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
或者在application.properties中添加 # application.properties
spring.application.name=Spring-boot-rabbitmq
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
简单的使用RabbitMQ
队列配置 这里得讲一下都代表什么，这是最基础的队列配置，配置了一个名为hello的队列 //RabbitConfig.java
@Configuration
public class RabbitConfig {
@Bean
public Queue Queue(){
return new Queue("hello");
}
}
若想添加多个队列，只需安装上面的模板添加队列就行，例如 //RabbitConfig.java
@Configuration
public class RabbitConfig {
@Bean
public Queue Queue(){
return new Queue("one");
}
@Bean
public Queue 方法名(){
return new Queue("队列名");
}
//或者
@Bean
public Queue 队列名(){
return new Queue("队列名");
}
}
发送者
所谓发送者，就是将队列消息发送的人[类/方法]，@Component注解用于将该类可以被Spring容器管理
//Person1Sender.java
@Component
public class Person1Sender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send(){
String context = "hello我是Person1";
System.out.println("Sender发出去的信息:"+context);
//这个的第一个参数在这里可以说是队列名，但是这个的全名叫routing_key[路由健],之后会讲到
this.rabbitTemplate.convertAndSend("one",context);
}
}
通过注入AmqpTemplate，让sender将消息发送的one队列,context就是发送的内容，呈键值对一般
this.rabbitTemplate.convertAndSend("one",context);
接收者 接收者可以接收队列发过来的消息,注解**@RabbitListener(queques=“队列名”)是为了监听此队列是否有消息，并通过注解@RabbitHandler**所对应的方法中加入形参来接收队列的消息 //Person2Receiver.java
@Component
@RabbitListener(queues = "one")
public class Person2Receiver {
@RabbitHandler
public void process(String one){
System.out.println("Person2接收到的信息:"+one);
}
}
测试 在Controller中，先通过注入Person1Sender对象，然后再测试控制类中的方法写上person1Sender.send(),其实就相当于new了Person1Sender对象，然后调用了Person1Sender里的send()方法，最后再浏览器输入localhost:8080查看控制台就知道效果了 //test.java
@Controller
public class testController {
@Autowired
Person1Sender person1Sender;
@RequestMapping("/")
@ResponseBody
public String test(){
person1Sender.send();
return "test";
}
}
效果：
多对多
顾名思义就是，实验一个发送者对多个接收者和多个发送者对多个接收者
一对多
修改一下 Person1sender.java //Person1Sender.java
@Component
public class Person1Sender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send(int i){
String context = "hello我是Person1---------------"+i;
this.rabbitTemplate.convertAndSend("one",context);
}
}
添加一个Person3Receiver.java，跟Person1Receiver.java简直是**的一模一样 @Component
@RabbitListener(queues = "one")
public class Person3Receiver {
@RabbitHandler
public void process(String msg){
System.out.println("Person3接收到的信息:"+msg);
}
}
再修改一下测试控制类 //testController.java
@Controller
public class testController {
@Autowired
Person1Sender person1Sender;
@RequestMapping("/")
@ResponseBody
public String test(){
for (int i = 0; i < 50; i++) {
person1Sender.send(i);
}
return "test";
}
}
运行之后，看下控制台 突然发现，person2和person3接收的消息都是一半一半的，都是你一个我一个 多对多
直接添加一个Person4Sender.java，简直跟**Person1Sender.java****的一模一样 //Person4Sender.java
@Component
public class Person4Sender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send(int i){
String context = "hello我是Person4---------------"+i;
this.rabbitTemplate.convertAndSend("one",context);
}
}
再修改一下测试控制类 //testController.java
@Controller
public class testController {
@Autowired
Person1Sender person1Sender;
@Autowired
Person4Sender person4Sender;
@RequestMapping("/")
@ResponseBody
public String test(){
for (int i = 0; i < 50; i++) {
person1Sender.send(i);
person4Sender.send(i);
}
return "test";
}
}
重新启动运行下看控制台 真**神奇，3接4的消息，2接1的消息，也是一半一半 总结：说明接收者无论怎样，都会均匀的接收消息
对于传对象的支持
不需要任何配置，直接将String换成Object就行
不同类型的Exchange
在RabbitMQ中，引入了两个概念
routing_key[路由键] 用于绑定队列，每个队列都对应有个路由键，必须是一串字符，用句号（ . ） 隔开，例如 topic.msg exchange[交换机] 交换机的主要功能是接收消息并且转发到绑定的队列，不会储存消息 交换机有四种类型
Direct[默认]：直接通过匹配所绑定routing_key的队列，再对所监听routing_key队列的方法进行调用，简单一点就是匹配队列名，然后调用监听队列名的方法，最简单的使用RabbitMQ Topic：通过给routing_key制定规则，可使用以下两个通配符 *表示一个词 #表示零个或多个词 例如： topic.msg.* [表示路由键必须是"topic.msg.随意一个词" =>(topic.msg.random)√ (topic.msg.a.b)×] 其实*可以省略不写 topic.msgs.# [表示路由键必须是以"topic.msgs"开头的所有路由键都行=>(topic.msgs.a.b)√ (topic.msgs)√ (topic.msg.random)×] Headers：也是根据规则匹配，但是自定义程度高，性能不怎么好，一般不会使用 Fanout：其词意思为"展开"，就是广播模式，不管设置了多少个路由键和交换机类型，直接将消息发送给绑定该交换机的全部队列 OKOK，了解完概念，直接上demo来理解
Topic
配置Topic类型交换机及所对应的队列 //TopicRabbitMQConfig.java
@Configuration
public class TopicRabbitMQConfig {
@Bean
public Queue queueMsg(){
return new Queue("msg");
}
@Bean
public Queue queueMsgs(){
return new Queue("msgs");
}
@Bean
public TopicExchange exchange(){
return new TopicExchange("exchange");
}
@Bean
public Binding BindingExchangeMsg(Queue queueMsg,TopicExchange exchange){
return BindingBuilder.bind(queueMsg).to(exchange).with("topic.msg");
}
@Bean
public Binding BindingExchangeMsgs(Queue queueMsgs,TopicExchange exchange){
return BindingBuilder.bind(queueMsgs).to(exchange).with("topic.#");
}
好像有点长，没事，一Bean一Bean讲解 首先，最收悉不过的添加队列，这里添加了两个队列分别为msg和msgs，其实这不重要，区分一下就行了 @Bean
public Queue queueMsg(){
return new Queue("msg");
}
@Bean
public Queue queueMsgs(){
return new Queue("msgs");
}
然后，将Topic类型引出来，并将交换机名取为exchange，最好方法名和交换机名一样，免得搞混 @Bean
public TopicExchange exchange(){
return new TopicExchange("exchange");
}
最后，将两个队列分别绑定到对应的路由键规则上，这里得注意几点，这里的形参的名字一定要和上面的方法名一样，因为再SpringBoot中，使用了@Bean注解 @Bean
public Binding BindingExchangeMsg(Queue queueMsg,TopicExchange exchange){
return BindingBuilder.bind(queueMsg).to(exchange).with("topic.msg");
}
@Bean
public Binding BindingExchangeMsgs(Queue queueMsgs,TopicExchange exchange){
return BindingBuilder.bind(queueMsgs).to(exchange).with("topic.#");
}
如果还看不懂，简化一下，这下你总看得懂了吧，方法名最好规范一点"BindingExchange队列名"，别到时候谁都看不懂你的代码 @Bean
public Binding 方法名(Queue 队列方法名,交换机类型 交换机名字){
return BindingBuilder.bind(队列方法名).to(交换机名字).with("路由键规则");
}
添加Person5Sender.java 这个sender可能有点不一样，特别是rabbitTemplate.convertAndSend
第一个值：交换机名字[exchange]
第二个值：路由键[routing_key]
第三个值：消息[context]
我这里设置了比较多的方法 //Person5Sender.java
@Component
public class Person5Sender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send1(){
String context = "hello我是Person5,RoutingKey为topic.msg";
this.rabbitTemplate.convertAndSend("exchange","topic.msg",context);
}
public void send2(){
String context = "hello我是Person5,RoutingKey为topic.msgs";
this.rabbitTemplate.convertAndSend("exchange","topic.msgs",context);
}
public void send3(){
String context = "hello我是Person5,RoutingKey为topic.random";
this.rabbitTemplate.convertAndSend("exchange","topic.random",context);
}
public void send4(){
String context = "hello我是Person5,RoutingKey为topic.";
this.rabbitTemplate.convertAndSend("exchange","topic",context);
}
为了方便，我单独拎出来一个方法**send1()**来分析一波
这个方法用于发送消息，context变量为消息 AmqpTemplate的convertAndSend方法的三个值：
第一个值：交换机名字[exchange]
第二个值：路由键[routing_key]
第三个值：消息[context]
这里主要是注意路由键 这里的路由键会经过Topic类型的交换机[exchange]，然后，交换机会通过匹配你所填写的路由键，并将消息放在匹配到的队列中 public void send1(){
String context = "hello我是Person5,RoutingKey为topic.msg";
this.rabbitTemplate.convertAndSend("exchange","topic.msg",context);
}
添加Receivers.java 这个类，我直接把接收者变成一个方法 //Receivers.java
@Component
public class Receivers {
@RabbitListener(queues = "msg")
@RabbitHandler
public void person6(String msg){
System.out.println("Person6接收到的信息:"+msg);
}
@RabbitListener(queues = "msgs")
@RabbitHandler
public void person7(String msg){
System.out.println("Person7接收到的信息:"+msg);
}
}
随便拎出来一个方法，分析一波
@RabbitListener注解：用于监听该队列的消息，需要搭配@RabbitHandler一起使用，queues的值为配置时的队列名@RabbitHandler注解：用于绑定监听到该队列消息后的方法方法的形参：方法的形参名字不一定要一样，但是要与监听到的消息的类型一样 @RabbitListener(queues = "msg")
@RabbitHandler0
public void person6(String msg){
System.out.println("Person6接收到的信息:"+msg);
}
修改testController.java //testController.java
@Controller
public class testController {
@Autowired
Person5Sender person5Sender;
@RequestMapping("/")
@ResponseBody
public String test() throws InterruptedException {
person5Sender.send1();
Thread.sleep(1000);
System.out.println("==================================");
person5Sender.send2();
Thread.sleep(1000);
System.out.println("==================================");
person5Sender.send3();
Thread.sleep(1000);
System.out.println("==================================");
person5Sender.send4();
return "test";
}
}
调用Person5Sender的四个方法，并加点延迟区分一下，可得结果↓ 其实者很明显了
第一组： 路由键为topic.msg，那就看看符不符合通配符的匹配 第一个队列路由键的规则：topic.msg，也就是说，发送者的路由键也必须是topic.msg才能将消息发送到msg队列，所以，msg队列有消息被存入，触发了Person6() 第二个队列路由键的规则：topic.#，也就是说，发送者的路由键只要是topic字母开头，.后面无论有还是没有，都能将消息发送到msgs队列，从而触发Person7() 第二、三、四组： 路由键都要么是topic.后面加一个词或多个词的，要么是只有topic的，只能符合topic.#这个规则，从而触发Person7() 总结：其实Topic类型也是挺好理解的，重点在于路由键规则的指定和匹配
Headers headers是自定义匹配规则类型，需要用到的对象比较多，所以性能比较低，一般不会使用 他的自定义匹配规则为设定一组键值对规则，然后发送者中也会设定一组键值对规则，当绑定了队列BindingBuilder.bind(queue).to(headersExchange)加上的是.whereAll(map).match()，那发送者必须全部正确匹配才会将消息发送到绑定的队列；而加上.whereAny(map).match()，那发送者所设置的键值对只需包含在配置文件里的键值对就行
Fanout
同样的先配置Fanout的交换机和用于测试的队列，创建一个FanoutRabbitMQConfig.java //FanoutRabbitMQConfig.java
@Configuration
public class FanoutRabbitMQConfig {
@Bean
public Queue queueRadio(){
return new Queue("radio");
}
@Bean
public Queue queueTV(){
return new Queue("tv");
}
@Bean
public Queue queuePhone(){
return new Queue("phone");
}
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("fanoutExchange");
}
@Bean
public Binding bindingExchangeRadio(Queue queueRadio,FanoutExchange fanoutExchange){
return BindingBuilder.bind(queueRadio).to(fanoutExchange);
}
@Bean
public Binding bindingExchangeTV(Queue queueTV,FanoutExchange fanoutExchange){
return BindingBuilder.bind(queueTV).to(fanoutExchange);
}
@Bean
public Binding bindingExchangePhone(Queue queuePhone,FanoutExchange fanoutExchange){
return BindingBuilder.bind(queuePhone).to(fanoutExchange);
}
}
这个应该都挺好懂得了吧，就是把radio、tv、phone队列都绑定到FanoutExchange上 添加发送者，FanoutSender.java，分别给三个方法随便取了三个不同的路由键 //FanoutSender.java
@Component
public class FanoutSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send1(){
String context = "我是FanoutSender,发送了send1,路由键为'abcdefg'";
this.rabbitTemplate.convertAndSend("fanoutExchange","abcdefg", context);
}
public void send2(){
String context = "我是FanoutSender,发送了send2,路由键为' '";
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}
public void send3(){
String context = "我是FanoutSender,发送了send3,路由键为'fanout'";
this.rabbitTemplate.convertAndSend("fanoutExchange","fanout", context);
}
}
添加接收者 FanoutReceiver.java，用于监听对应队列的消息 //FanoutReceiver.java
@Component
public class FanoutReceiver {
@RabbitListener(queues = "radio")
@RabbitHandler
public void radio(String msg){
System.out.println("监听radio："+msg);
}
@RabbitListener(queues = "phone")
@RabbitHandler
public void phone(String msg){
System.out.println("监听phone："+msg);
}
@RabbitListener(queues = "tv")
@RabbitHandler
public void tv(String msg){
System.out.println("监听tv："+msg);
}
}
修改测试控制类testController.java //testController.java
@Controller
public class testController {
@Autowired
FanoutSender fanoutSender;
@RequestMapping("/")
@ResponseBody
public String test() throws InterruptedException {
fanoutSender.send1();
Thread.sleep(1000);
System.out.println("==========================");
fanoutSender.send2();
Thread.sleep(1000);
System.out.println("==========================");
fanoutSender.send3();
return "test";
}
}
查看效果 总结：很明显了，这三个队列无论发送者是什么路由键，他都会将消息发送到三个队列中
demo
GitHub Gitee
参考文档：
Spring Boot3.0(八)：消息队列 RabbitMQ 详解 - 纯洁的微笑博客 (ityouknow.com)
相关链接
发表评论