消息可靠性
生产者确认机制
提出问题:消息投递过程中,生产者——> MQ ——> 消费者 中间会出现消息丢失问题,导致信息没有及时同步
先梳理一下流程
生产者生产个消息 ——> 建立连接——>通道传递进mq交换机——>交换机传给队列——>消费者拉取数据消费
1.生产者生产完消息,相当于写好代码,写错了自己改,然后建立连接投递,连接建立不成功会再建立,这里不用操心,如果在投递过程中消息丢失了,生产者发送了,消费者没收到,这要是情侣铁定闹分手,有可能是网络波动造成的也有可能没有给了一个不存在的交换机,所以mq官方需要处理这些问题,这就交给他们了,但是我们要了解其中的处理方案
发现问题:消息没到交换机或者到了交换机没到队列,这两种情况基本一致,因为交换机没有存储信息能力
理论解决方案:没到mq的队列就丢失了,连给队列持久化保存到硬盘的机会都没得,那就只能重新传一份。
mq的实践解决方案:生产者确认机制,首先需要给每个信息绑定全局唯一的ID,信息可能有上百万条,谁知道哪个丢失了,然后生产者发生了一个带有ID的信息,连接一旦建立,交换机这边是否接收到了这条ID,接收到了给生产者发一个ack,让他先别发了,没接收到也得发一个nack,如果路由到队列后,则发生一个回执ack,说我任务完成了
消息持久化
前面的生产者确认机制已经确保消息可靠的传输到队列,如果RabbitMQ宕机的话,依旧会导致小心丢失,由于MQ是内存存储,所以需要持久化保存到硬盘
这里面涉及到了三个持久化
交换机持久化,队列持久化,消息持久化
1.交换机默认是非持久化的,mq重启就会丢失,SpringAMQP中可以通过代码指定交换机持久化,默认情况下,由SpringAMQP声明的交换机都是持久化的。
@Bean public DirectExchange simpleExchange(){ // 三个参数:交换机名称、是否持久化、当没有queue与其绑定时是否自动删除 return new DirectExchange("simple.direct", true, false); }
2.队列持久化
队列默认是非持久化的,mq重启就会丢失,SpringAMQP中可以通过代码指定队列持久化,默认情况下,由SpringAMQP声明的队列都是持久化的。
@Bean public Queue simpleQueue(){ // 使用QueueBuilder构建队列,durable就是持久化的 return QueueBuilder.durable("simple.queue").build(); }
3.消息持久化
消息持久化是把生产者生产的信息需要持久化,保存到硬盘里
默认情况下,SpringAMQP发出的任何消息都是持久化的,不用特意指定。
消费者消息确认
RabbitMQ是阅后即焚机制,RabbitMQ确认消息被消费者消费后会立刻删除。
而RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的:消费者获取消息后,应该向RabbitMQ发送ACK回执,表明自己已经处理消息。
这里面涉及一点,消息是存储在MQ当中,消费者拉取完之后,MQ删除消息的时机是哪,如果消费者拉取完之后,没消费呢,自己挂了,又怎么办
结合以上两点,最可靠的就是消费者消费完了,通知一声我完事了,然后MQ把消息删除,防止占用内存
MQ提供了三种机制
- manual:手动ack,需要在业务代码结束后,调用api发送ack。
- auto:自动ack,由spring监测listener代码是否出现异常,没有异常则返回ack;抛出异常则返回nack
- none:关闭ack,MQ假定消费者获取消息后会成功处理,因此消息投递后立即被删除
很显然none模式下,消费者获取完消息之后,服务一旦宕机,消息也就丢失了,除非能确保自己的服务非常稳定才使用
auto模式类似于事务的回滚,消息一旦长时间没被消费会抛出异常,导致回滚,MQ消息不会删除
消费失败重试机制
当消费者出现异常后,消息会不断requeue(重入队)到队列,再重新发送给消费者,然后再次异常,再次requeue,无限循环,导致mq的消息处理飙升,带来不必要的压力:
解决方案:本地重试
可以利用Spring的retry机制,在消费者出现异常时利用本地重试,而不是无限制的requeue到mq队列。
修改consumer服务的application.yml文件,添加内容:
spring: rabbitmq: listener: simple: retry: enabled: true # 开启消费者失败重试 initial-interval: 1000 # 初识的失败等待时长为1秒 multiplier: 1 # 失败的等待时长倍数,下次等待时长 = multiplier * last-interval max-attempts: 3 # 最大重试次数 stateless: true # true无状态;false有状态。如果业务中包含事务,这里改为false
重启消费者服务,重复之前的测试。可以发现:
在重试3次后,SpringAMQP会抛出异常AmqpRejectAndDontRequeueException,说明本地重试触发了
查看RabbitMQ控制台,发现消息被删除了,说明最后SpringAMQP返回的是ack,mq删除消息了
结论:
开启本地重试时,消息处理过程中抛出异常,不会requeue到队列,而是在消费者本地重试
重试达到最大次数后,Spring会返回ack,消息会被丢弃
失败策略
在之前的测试中,达到最大重试次数后,消息会被丢弃,这是由Spring内部机制决定的。
在开启重试模式后,重试次数耗尽,如果消息依然失败,则需要有MessageRecovery接口来处理,它包含三种不同的实现:
RejectAndDontRequeueRecoverer:重试耗尽后,直接reject,丢弃消息。默认就是这种方式
ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队
RepublishMessageRecoverer:重试耗尽后,将失败消息投递到指定的交换机
比较优雅的一种处理方案是RepublishMessageRecoverer,失败后将消息投递到一个指定的,专门存放异常消息的队列,后续由人工集中处理。这个队列就是死信队列
Rabbitmq的可靠性保证
通过以下四点保证
开启生产者确认机制,确保生产者的消息能到达队列
开启持久化功能,确保消息未消费前在队列中不会丢失
开启消费者确认机制为auto,由spring确认消息处理成功后完成ack文章来源:https://www.toymoban.com/news/detail-843005.html
开启消费者失败重试机制,并设置MessageRecoverer,多次重试失败后将消息投递到异常交换机,交由人工处理文章来源地址https://www.toymoban.com/news/detail-843005.html
到了这里,关于RabbitMQ服务异步通信-高级篇的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!