持久化
消息的可靠性是 RabbitMQ 的一大特色,那么 RabbitMQ 是如何保证消息可靠性的呢——消息持久化。
创建两个项目
服务消费者
服务提供者
我们在提供者处每隔两秒发送一个消息到交换器中,如下
@RunWith(SpringRunner.class) @SpringBootTest(classes = RabbitmqDirectProviderApplication.class) public class RabbitmqDirectProviderApplicationTests { @Autowired private Sender sender; @Test public void contextLoads() throws Exception{ int flag = 0; while(true){ Thread.sleep(2000); sender.send("hello .... "+flag++); } } }
启动消费者和服务提供者
当我们在15的时候关闭消费者,然后在启动消费者我们看情况。
我们会发现在消费者下线的这段时间产生的消息并没有丢失
autoDelete 属性
@Queue: 当所有消费客户端连接断开后,是否自动删除队列
true:删除 false:不删除
@Exchange:当所有绑定队列都不在使用时,是否自动删除交换器
true:删除 false:不删除
ACK确认机制
1.什么是消息确认ACK?
如果在处理消息的过程中,消费者的服务器在处理消息时出现异常,那么可能这条正在处理的消息刘没有完成消息消费,数据就会丢失,为了确保数据不会丢失RabbitMQ支持消息确认-ACK
2.ACK的消息确认机制
ACK机制是消费者从RabbitMQ收到消息并处理完成后,反馈给RabbitMQ的,RabbitMQ收到反馈后才将此消息从队列中删除。
如果一个消费者在处理消息出现了网络不稳定、服务器异常等现象,会将消息重新放入队列中。
如果在集群的情况下,RabbitMQ会立即将这个消息推送给这个在线的其他消费者,这种机制保证了在消费者服务端故障的时候,不会丢失任何消息和任务。
消息永远不会从RabbitMQ中删除:只有当消费者正确发送ACK反馈,RabbitMQ确认收到后,消息才会从RabbitMQ服务器的数据中删除。
消息的ACK确认机制默认是打开的
演示ACK场景,消费的异常
消费者在一直弹出异常信息,说明消息一直被重复的添加到队列中
3.ACK机制的开发注意事项
如果忘记ACK,那么后果会很严重,当Consumer退出时,Message会一直重新分发,然后RabbitMQ会占用越来越多的内存,由于RabbitMQ会长时间运行,因此这个“内存泄漏”是致命的。我们可以通过设置重试的次数要防止这个问题
在consumer的application.properties中添加如下配置
spring.rabbitmq.listener.direct.retry.enabled=true # 重试次数 默认为3次 spring.rabbitmq.listener.direct.retry.max-attempts=5