1、RabbitMQ的消息持久化处理&＃xff0c;消息的可靠性是 RabbitMQ 的一大特色&＃xff0c;那么 RabbitMQ 是如何保证消息可靠性的呢——消息持久化。
2、autoDelete属性的理解。
1&＃xff09;、&＃64;Queue: 当autoDelete属性设置到该注解的时候&＃xff0c;含义即是&＃xff0c;当所有消费者客户端连接断开后&＃xff0c;是否自动删除队列&＃xff0c;当设置值是true的时候删除该队列&＃xff0c;当值是false的时候不删除该队列。
2&＃xff09;、&＃64;Exchange&＃xff1a;当autoDelete属性设置到该注解的时候&＃xff0c;含义即是&＃xff0c;当所有绑定队列都不在使用时&＃xff0c;是否自动删除交换器&＃xff0c;当设置值是true的时候删除该交换器&＃xff0c;当值是false的时候不删除该交换器。
3、之前写过RabbitMQ的交换器Exchange之direct&＃xff08;发布与订阅 完全匹配&＃xff09;&＃xff0c;这里借助这个进行消息持久化测试。生产者的代码不发生改变&＃xff0c;这里只是将消费者的autoDelete &＃61; "true"属性修改为autoDelete &＃61; "false"&＃xff0c;进行对比测试。
Info级别的日志进行消息的持久化操作&＃xff0c;即队列不进行自动删除。将autoDelete &＃61; "false"即可。
1 package com.example.bie.consumer;
2
3 import org.springframework.amqp.core.ExchangeTypes;
4 import org.springframework.amqp.rabbit.annotation.Exchange;
5 import org.springframework.amqp.rabbit.annotation.Queue;
6 import org.springframework.amqp.rabbit.annotation.QueueBinding;
7 import org.springframework.amqp.rabbit.annotation.RabbitHandler;
8 import org.springframework.amqp.rabbit.annotation.RabbitListener;
9 import org.springframework.stereotype.Component;
10
11 /**
12 *
13 * &＃64;author biehl
14 *
15 * 消息接收者
16 *
17 * 1、&＃64;RabbitListener bindings:绑定队列
18 *
19 * 2、&＃64;QueueBinding
20 * value:绑定队列的名称、exchange:配置交换器、key:路由键routing-key绑定队列和交换器
21 *
22 * 3、&＃64;Queue value:配置队列名称、autoDelete:是否是一个可删除的临时队列
23 *
24 * 4、&＃64;Exchange value:为交换器起个名称、type:指定具体的交换器类型
25 *
26 *
27 */
28 &＃64;Component
29 &＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(
30
31 value &＃61; &＃64;Queue(value &＃61; "${rabbitmq.config.queue.info}", autoDelete &＃61; "false"),
32
33 exchange &＃61; &＃64;Exchange(value &＃61; "${rabbitmq.config.exchange}", type &＃61; ExchangeTypes.DIRECT),
34
35 key &＃61; "${rabbitmq.config.queue.info.routing.key}"))
36 public class LogInfoConsumer {
37
38 /**
39 * 接收消息的方法,采用消息队列监听机制.
40 *
41 * &＃64;RabbitHandler意思是将注解&＃64;RabbitListener配置到类上面
42 *
43 * &＃64;RabbitHandler是指定这个方法可以进行消息的接收并且消费.
44 *
45 * &＃64;param msg
46 */
47 &＃64;RabbitHandler
48 public void consumer(String msg) {
49 // 打印消息
50 System.out.println("INFO消费者&＃61;&＃61;&＃61;>消费: " &＃43; msg);
51 }
52
53 }
Error级别的日志进行消息的持久化操作&＃xff0c;即队列进行自动删除。将autoDelete &＃61; "true"即可。
1 package com.example.bie.consumer;
2
3 import org.springframework.amqp.core.ExchangeTypes;
4 import org.springframework.amqp.rabbit.annotation.Exchange;
5 import org.springframework.amqp.rabbit.annotation.Queue;
6 import org.springframework.amqp.rabbit.annotation.QueueBinding;
7 import org.springframework.amqp.rabbit.annotation.RabbitHandler;
8 import org.springframework.amqp.rabbit.annotation.RabbitListener;
9 import org.springframework.stereotype.Component;
10
11 /**
12 *
13 * &＃64;author biehl
14 *
15 * 消息接收者
16 *
17 * 1、&＃64;RabbitListener bindings:绑定队列
18 *
19 * 2、&＃64;QueueBinding
20 * value:绑定队列的名称、exchange:配置交换器、key:路由键routing-key绑定队列和交换器
21 *
22 * 3、&＃64;Queue value:配置队列名称、autoDelete:是否是一个可删除的临时队列
23 *
24 * 4、&＃64;Exchange value:为交换器起个名称、type:指定具体的交换器类型
25 *
26 *
27 */
28 &＃64;Component
29 &＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(
30
31 value &＃61; &＃64;Queue(value &＃61; "${rabbitmq.config.queue.error}", autoDelete &＃61; "true"),
32
33 exchange &＃61; &＃64;Exchange(value &＃61; "${rabbitmq.config.exchange}", type &＃61; ExchangeTypes.DIRECT),
34
35 key &＃61; "${rabbitmq.config.queue.error.routing.key}"))
36 public class LogErrorConsumer {
37
38 /**
39 * 接收消息的方法,采用消息队列监听机制.
40 *
41 * &＃64;RabbitHandler意思是将注解&＃64;RabbitListener配置到类上面
42 *
43 * &＃64;RabbitHandler是指定这个方法可以进行消息的接收并且消费.
44 *
45 * &＃64;param msg
46 */
47 &＃64;RabbitHandler
48 public void consumer(String msg) {
49 // 打印消息
50 System.out.println("ERROR消费者&＃61;&＃61;&＃61;>消费<&＃61;&＃61;&＃61;消息message: " &＃43; msg);
51 }
52
53 }
4、启动你的生产者&＃xff0c;启动你的消费者&＃xff0c;观察RabbitMQ的图形化界面。未生产消息、未消费消息的界面如下所示&＃xff1a;
生产消息、消费消息的界面如下所示&＃xff0c;我这里还使用浏览器访问控制层触发生产者生产消息&＃xff0c;消费者消费消息&＃xff1a;
现在停止你的消费者&＃xff0c;记录消息到第几条消息了。方便再次启动消费者进行观察。
启动你的消费者&＃xff0c;观察&＃xff0c;看看是从第几条开始消费的。可以看到消息从第82条开始消费的。
RabbitMQ的消息持久化处理&＃xff0c;Ready是对未接收到的数据状态表示&＃xff0c;如果RabbitMQ在队列里面存放的消息未被消费者所消费&＃xff0c;那么会给未消费的消息加一个标记&＃xff0c;表示当前这个消息未被消费。消息持久化处理解决了丢失消息的这种状况&＃xff0c;我们可以接收到消息&＃xff0c;就是因为队列一直存在着呢&＃xff0c;但是手动删除队列&＃xff0c;消息也就丢失了&＃xff0c;所以要慎重操作。当消费者停止以后&＃xff0c;生产者生产的消息存储在RabbitMQ的服务器内存中&＃xff0c;队列也存在内存中&＃xff0c;数据在队列中&＃xff0c;即数据保存在内存中。但是如果RabbitMQ的服务都停止了&＃xff0c;队列也就消失了&＃xff0c;队列消失了&＃xff0c;数据也就丢失了。
作者&＃xff1a;别先生
博客园&＃xff1a;https://www.cnblogs.com/biehongli/
如果您想及时得到个人撰写文章以及著作的消息推送&＃xff0c;可以扫描上方二维码&＃xff0c;关注个人公众号哦。
京公网安备 11010802041100号