SpringBoot+RabbitMQ学习笔记(五)RabbitMQ消息…
2020-04-29 16:08:54来源:博客园 阅读 ()
SpringBoot+RabbitMQ学习笔记(五)RabbitMQ消息持久化处理
一丶简介
在@Queue和@Exchange注解中都有autoDelete属性,值是布尔类型的字符串。如:autoDelete=“false”。
@Queue:当所有消费客户端断开连接后,是否自动删除队列: true:删除,false:不删除。
@Exchange:当所有绑定队列都不在使用时,是否自动删除交换器: true:删除,false:不删除。
当所有消费客户端断开连接时,而我们对RabbitMQ消息进行了持久化,那么这时未被消费的消息存于RabbitMQ服务器的内存中,如果RabbitMQ服务器都关闭了,那么未被消费的数据也都会丢失了。
下面编写代码试试RabbitMQ的消息持久化处理。
二丶配置文件
这里使用的是前面博客里写的error日志消息队列来测试的。这里还是创建两个项目,一个作为生产者,一个作为消费者。
生产者配置:
server.port=8883 spring.application.name=hello-world spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest spring.thymeleaf.cache=false #设置交换器名称 mq.config.exchange=log.direct #设置error队列的路由键 mq.config.queue.error.routing.key=log.error.routing.keyView Code
消费者配置
server.port=8884 spring.application.name=lesson1 spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest #设置交换器名称 mq.config.exchange=log.direct #设置error队列名称 mq.config.queue.error=log.error #设置error路由键 mq.config.queue.error.routing.key=log.error.routing.keyView Code
三丶编写生产者
package com.example.rabbitdurableprovider; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; /** * Author:aijiaxiang * Date:2020/4/26 * Description:发送消息 */ @Component public class DurableSender { @Autowired private AmqpTemplate amqpTemplate; //exChange 交换器 @Value("${mq.config.exchange}") private String exChange; //routingkey 路由键 @Value("${mq.config.queue.error.routing.key}") private String routingKey; /** * 发送消息的方法 * @param msg */ public void send(String msg){ //向消息队列发送消息 //参数1:交换器名称 //参数2:路由键 //参数3:消息 this.amqpTemplate.convertAndSend(exChange,routingKey,msg); } }View Code
四丶编写消费者
这里消费者服务配置中@Queue中的autoDelete属性设置的是true,即未持久化,一会儿测试下看没有持久化的消息队列在所有的消费者服务器断开后是怎样的。
package com.ant.rabbitdurableconsumer; import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.rabbit.annotation.*; import org.springframework.stereotype.Component; /** * Author:aijiaxiang * Date:2020/4/26 * Description:消息接收者 * @RabbitListener bindings:绑定队列 * @QueueBinding value:绑定队列的名称 * exchange:配置交换器 * @Queue : value:配置队列名称 * autoDelete:是否是一个可删除的临时队列 * @Exchange value:为交换器起个名称 * type:指定具体的交换器类型 */ @Component @RabbitListener( bindings = @QueueBinding( value = @Queue(value = "${mq.config.queue.error}",autoDelete = "true"), exchange = @Exchange(value = "${mq.config.exchange}", type = ExchangeTypes.DIRECT), key = "${mq.config.queue.error.routing.key}" ) ) public class DurableErrorReceiver { /** * 接收消息的方法,采用消息队列监听机制 * @param msg */ @RabbitHandler public void process(String msg){ System.out.println("error-receiver:"+msg); } }View Code
五丶编写测试类
这里写了一个死循环持续向消息队列中发送消息,用变量falg来记录发送编号。
package com.example.amqp; import com.example.helloworld.HelloworldApplication; import com.example.rabbitdurableprovider.DurableSender; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; /** * Author:aijiaxiang * Date:2020/4/26 * Description: */ @RunWith(SpringRunner.class) @SpringBootTest(classes = HelloworldApplication.class) public class QueueTest { @Autowired private DurableSender durableSender; @Test public void test3() throws InterruptedException { int flag = 0; while (true){ flag++; Thread.sleep(2000); durableSender.send("hello--"+flag); } } }View Code
先启动消费者服务器,然后启动测试类,控制台输出如下信息,然后关闭tomcat,模拟消费者服务器故障。这里可以看到消费者接受到的消息停留在第81条就出现了“故障”,但是生产者还是在持续不断的向消费者发送消息。
这时再重启tomcat,消费者又接受到了消息,但是是从第111条消息开始的,那么81-111之间的这些消息就都丢失了。
修改消费者服务的代码,将autoDelete设置为“false”,将RabbitMQ消息进行持久化处理。
@RabbitListener( bindings = @QueueBinding( value = @Queue(value = "${mq.config.queue.error}",autoDelete = "false"), exchange = @Exchange(value = "${mq.config.exchange}", type = ExchangeTypes.DIRECT), key = "${mq.config.queue.error.routing.key}" ) )
修改后重启消费者服务器,再次调用测试方法。然后关闭消费者服务器,模拟“故障”。这时看到消息接收到第15条服务器就“故障”了。
重启消费者服务器。可以看到服务器一启动,消费者就从消息队列中读取到了服务器“故障”时缓存在RabbitMQ中的消息,消息并未丢失,RabbitMQ消息持久化处理成功。
OK!以上就是今天学习的RabbitMQ消息持久化处理,如有不对之处,欢迎指正!
原文链接:https://www.cnblogs.com/aijiaxiang/p/12802893.html
如有疑问请与原作者联系
标签:
版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有
- 学习Java 8 Stream Api (4) - Stream 终端操作之 collect 2020-06-11
- java学习之第一天 2020-06-11
- Java学习之第二天 2020-06-11
- Spring WebFlux 学习笔记 - (一) 前传:学习Java 8 Stream Ap 2020-06-11
- Java笔记:集合 2020-06-10
IDC资讯: 主机资讯 注册资讯 托管资讯 vps资讯 网站建设
网站运营: 建站经验 策划盈利 搜索优化 网站推广 免费资源
网络编程: Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它
服务器技术: Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护
软件技巧: 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷
网页制作: FrontPages Dreamweaver Javascript css photoshop fireworks Flash