SpringBoot之RabbitMQ的使用
一 、RabbitMQ的介绍
RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件,消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下:
从上图可看出,对于消息队列来说,生产者、消息队列、消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息队列,并且当消息队列收到消息之后,接收消息队列传来的消息,并且给予相应的处理。消息队列常用于分布式系统之间互相信息的传递。
对于RabbitMQ来说,除了这三个基本模块以外,还添加了一个模块,即交换机(Exchange)。它使得生产者和消息队列之间产生了隔离,生产者将消息发送给交换机,而交换机则根据调度策略把相应的消息转发给对应的消息队列。
交换机的主要作用是接收相应的消息并且绑定到指定的队列。交换机有四种类型,分别为Direct、topic、headers、Fanout。
Direct是RabbitMQ默认的交换机模式,也是最简单的模式。即创建消息队列的时候,指定一个BindingKey。当发送者发送消息的时候,指定对应的Key。当Key和消息队列的BindingKey一致的时候,消息将会被发送到该消息队列中。
topic转发信息主要是依据通配符,队列和交换机的绑定主要是依据一种模式(通配符+字符串),而当发送消息的时候,只有指定的Key和该模式相匹配的时候,消息才会被发送到该消息队列中。
headers也是根据一个规则进行匹配,在消息队列和交换机绑定的时候会指定一组键值对规则,而发送消息的时候也会指定一组键值对规则,当两组键值对规则相匹配的时候,消息会被发送到匹配的消息队列中。
Fanout是路由广播的形式,将会把消息发给绑定它的全部队列,即便设置了key,也会被忽略。
二 、SpringBoot整合RabbitMQ(Direct模式)
SpringBoot整合RabbitMQ非常简单,首先还是pom.xml引入依赖。
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>
在application.properties中配置RabbitMQ相关的信息,并首先启动了RabbitMQ实例,并创建两个queue。
spring.application.name=spirng-boot-rabbitmq spring.rabbitmq.host=127.0.0.1 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=admin
配置Queue(消息队列),由于采用的是Direct模式,需要在配置Queue的时候指定一个键,使其和交换机绑定。
@Configuration public class RabbitConfig {@Beanpublic org.springframework.amqp.core.Queue Queue() {return new org.springframework.amqp.core.Queue("hello");} }
接着就可以发送消息啦。在SpringBoot中,我们使用AmqpTemplate去发送消息。代码如下:
@Component public class HelloSender {@Autowiredprivate AmqpTemplate rabbitTemplate;public void send(int index) {String context = "hello Queue "+index + new Date();System.out.println("Sender : " + context);this.rabbitTemplate.convertAndSend("hello", context);} }
生产者发送消息之后就需要消费者接收消息。这里定义了两个消息消费者,用来模拟生产者与消费者一对多的关系。
@Component @RabbitListener(queues = "hello") public class HelloReceiver {@RabbitHandlerpublic void process(String hello) {System.out.println("Receiver1 : " + hello);} } @Component @RabbitListener(queues = "hello") public class HelloReceiver2 {@RabbitHandlerpublic void process(String hello) {System.out.println("Receiver2 : " + hello);} }
在单元测试中模拟发送消息,批量发送10条消息,两个接收者分别接收了5条消息。
@Autowiredprivate HelloSender helloSender;@Testpublic void hello() throws Exception {for(int i=0;i<10;i++){helloSender.send(i);}}
实际上RabbitMQ还可以支持发送对象,当然由于涉及到序列化和反序列化,该对象要实现Serilizable接口。这里定义了User对象,用来做发送消息内容。
import java.io.Serializable; public class User implements Serializable{private String name;private String pwd;public String getPwd() {return pwd;}public void setPwd(String pwd) {this.pwd = pwd;}public String getName() {return name;}public void setName(String name) {this.name = name;}public User(String name, String pwd) {this.name = name;this.pwd = pwd;} @Override public String toString() {return "User{" +"name='" + name + '\'' +", pwd='" + pwd + '\'' +'}';} }
在生产者中发送User对象。
@Component public class ModelSender {@Autowiredprivate AmqpTemplate rabbitTemplate;public void sendModel(User user) {System.out.println("Sender object: " + user.toString());this.rabbitTemplate.convertAndSend("object", user);} }
在消费者中接收User对象。
@Component @RabbitListener(queues = "object") public class ModelRecevicer {@RabbitHandlerpublic void process(User user) {System.out.println("Receiver object : " + user);} }
在单元测试中注入ModelSender 对象,实例化User对象,然后发送。
@Autowiredprivate ModelSender modelSender;@Testpublic void model() throws Exception {User user=new User("abc","123"); modelSender.sendModel(user);}
三 、SpringBoot整合RabbitMQ(Topic转发模式)
首先需要在RabbitMQ服务端创建交换机topicExchange,并绑定两个queue:topic.message、topic.messages。
新建TopicRabbitConfig,设置对应的queue与binding。
@Configuration public class TopicRabbitConfig {final static String message = "topic.message";final static String messages = "topic.messages";@Beanpublic Queue queueMessage() {return new Queue(TopicRabbitConfig.message);} @Beanpublic Queue queueMessages() {return new Queue(TopicRabbitConfig.messages);}@BeanTopicExchange exchange() {return new TopicExchange("topicExchange");}@BeanBinding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");}@BeanBinding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");} }
创建消息生产者,在TopicSender中发送3个消息。
@Component public class TopicSender {@Autowiredprivate AmqpTemplate rabbitTemplate;public void send() {String context = "hi, i am message all";System.out.println("Sender : " + context);this.rabbitTemplate.convertAndSend("topicExchange", "topic.1", context);}public void send1() {String context = "hi, i am message 1";System.out.println("Sender : " + context);this.rabbitTemplate.convertAndSend("topicExchange", "topic.message", context);}public void send2() { String context = "hi, i am messages 2";System.out.println("Sender : " + context);this.rabbitTemplate.convertAndSend("topicExchange", "topic.messages", context); } }
生产者发送消息,这里创建了两个接收消息的消费者。
@Component @RabbitListener(queues = "topic.message") public class TopicReceiver {@RabbitHandlerpublic void process(String message) {System.out.println("Topic Receiver1 : " + message);} }@Component @RabbitListener(queues = "topic.messages") public class TopicReceiver2 {@RabbitHandlerpublic void process(String message) {System.out.println("Topic Receiver2 : " + message);} }
在单元测试中注入TopicSender,利用topicSender 发送消息。
@Autowiredprivate TopicSender topicSender;@Testpublic void topicSender() throws Exception { topicSender.send(); topicSender.send1(); topicSender.send2();}
从上面的输出结果可以看到,Topic Receiver2 匹配到了所有消息,Topic Receiver1只匹配到了1个消息。
四 、SpringBoot整合RabbitMQ(Fanout Exchange形式)
Fanout Exchange形式又叫广播形式,因此我们发送到路由器的消息会使得绑定到该路由器的每一个Queue接收到消息。首先需要在RabbitMQ服务端创建交换机fanoutExchange,并绑定三个queue:fanout.A、fanout.B、fanout.C。
与Topic类似,新建FanoutRabbitConfig,绑定交换机和队列。
@Configuration public class FanoutRabbitConfig {@Beanpublic Queue AMessage() {return new Queue("fanout.A");}@Beanpublic Queue BMessage() {return new Queue("fanout.B");}@Beanpublic Queue CMessage() {return new Queue("fanout.C");}@BeanFanoutExchange fanoutExchange() {return new FanoutExchange("fanoutExchange");}@BeanBinding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {return BindingBuilder.bind(AMessage).to(fanoutExchange);}@BeanBinding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {return BindingBuilder.bind(BMessage).to(fanoutExchange);}@BeanBinding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {return BindingBuilder.bind(CMessage).to(fanoutExchange);} }
创建消息生产者,在FanoutSender中发送消息。
@Component public class FanoutSender {@Autowiredprivate AmqpTemplate rabbitTemplate;public void send() {String context = "hi, fanout msg ";System.out.println("FanoutSender : " + context);this.rabbitTemplate.convertAndSend("fanoutExchange","", context);} }
然后创建了3个接收者FanoutReceiverA、FanoutReceiverB、FanoutReceiverC。
@Component @RabbitListener(queues = "fanout.A") public class FanoutReceiverA {@RabbitHandlerpublic void process(String message) {System.out.println("fanout Receiver A : " + message);} } @Component @RabbitListener(queues = "fanout.B") public class FanoutReceiverB {@RabbitHandlerpublic void process(String message) {System.out.println("fanout Receiver B: " + message);} } @Component @RabbitListener(queues = "fanout.C") public class FanoutReceiverC {@RabbitHandlerpublic void process(String message) {System.out.println("fanout Receiver C: " + message);} }
在单元测试中注入消息发送者,发送消息。
@Autowiredprivate FanoutSender fanoutSender;@Testpublic void fanoutSender() throws Exception {fanoutSender.send();}
从下图可以看到3个队列都接收到了消息。
本章节创建的类比较多,下图为本章节的结构,也可以直接查看demo源码了解。
SpringBoot之RabbitMQ的使用相关推荐
- RabbitMQ(八):SpringBoot 整合 RabbitMQ(三种消息确认机制以及消费端限流)
说明 本文 SpringBoot 与 RabbitMQ 进行整合的时候,包含了三种消息的确认模式,如果查询详细的确认模式设置,请阅读:RabbitMQ的三种消息确认模式 同时消费端也采取了限流的措施, ...
- 九、springboot整合rabbitMQ
springboot整合rabbitMQ 简介 rabbitMQ是部署最广泛的开源消息代理. rabbitMQ轻量级,易于在内部和云中部署. 它支持多种消息传递协议. RabbitMQ可以部署在分布式 ...
- SpringBoot使用RabbitMQ消息队列
RabbitMQ简介 AMQP,即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计.消息中间件主要用于组件之间的 ...
- springboot 集成rabbitmq 实例
springboot 集成rabbitmq 实例 个人在学习rabbitmq时发现网上很少有系统性介绍springboot和rabbitmq如何集成的,其他人总结的都片段化,所以结合个人调研过程,整理 ...
- RabbitMQ,RabbitMQ 的工作模式,Spring 整合 RabbitMQ,Springboot 整合RabbitMQ
什么是RabbitMQ 1.1 MQ概述 MQ全称 Message Queue(消息队列),是在消息的传输过程中保存消息的容器.多用于分布式系统之间进行通信. ⚫ MQ,消息队列,存储消息的中间件 ⚫ ...
- SpringBoot整合 ActiveMQ、SpringBoot整合RabbitMQ、SpringBoot整合Kafka
1.概念:SpringBoot 整合消息服务2.具体内容对于异步消息组件在实际的应用之中会有两类:· JMS:代表作就是 ActiveMQ,但是其性能不高,因为其是用 java 程序实现的:· AMQ ...
- RabbitMq(九) SpringBoot整合RabbitMQ消费者示例代码
概述 在上一篇我们介绍了SpringBoot整合RabbitMQ生产者代码,本章我们介绍SpringBoot整合RabbitMQ,实现消费者工程的代码实现.与生产者集成相比,集成消费者不需要进行添加配 ...
- RabbitMq(八) SpringBoot整合RabbitMQ 生产者代码实现
在本章中我们将创建RabbitMQ的生产者工程,并实现生产者端代码实现. springboot整合RabbitMQ生产者工程步骤如下: 创建maven工程 引入springboot及RabbitMQ依 ...
- springboot之rabbitmq
一.RabbitMQ是实现了高级消息队列协议(AMQP)的开源消息代理软件(亦称面向消息的中间件).RabbitMQ服务器是用Erlang语言编写的,而集群和故障转移是构建在开放电信平台框架上的.所有 ...
- SpringBoot和RabbitMQ集成
SpringBoot和RabbitMQ的集成: 步骤 自动配置 123456789101112131415161718192021222324252627 @Bean public CachingCo ...
最新文章
- 共享可写节包含重定位_PE结构学习01-DOS头-NT头-节表头
- php绑定变量,php动态绑定变量的用法
- C# 视频监控系列(14):总结贴——VC++代码转成C#小结
- jquery.cookie 使用方法
- mysql周报内容范文_Mysql各种表格查询含实例,日报,周报,月报,时间差自动计算...
- Android小项目之--服务【Service】
- (1.1)HarmonyOS鸿蒙中Ability概念及意义
- 解释性语言和汇编性语言对比
- 从0到1写RT-Thread内核——临界段的保护
- IUS database
- iOS与JS交互之UIWebView协议拦截
- 电动汽车动力系统整车仿真模型,具有双向DCDC变换器实现能量反馈,带异步电机仿真,应用最大转矩电流比控制加独特的弱磁控制策略
- js实现跟浏览器 ctrl+f 一样的效果
- 万字专栏总结 | 离线强化学习(OfflineRL)总结(原理、数据集、算法、复杂性分析、超参数调优等)...
- 不要说话 -- 陈奕迅/小柯
- 微型计算机蓝屏的处理方法,电脑出现蓝屏如何解决_电脑蓝屏的处理方法
- CTF解题基本思路步骤(misc和web)
- 海量数据处理算法 各种STL容器使用的数据结构剖析
- 【Prefix-Tuning】Fine-tuning过时啦,NLG生成任务如何Tuning?
- Java常见面试50题(java jsp)