网站开发合同 附件,免费生成ppt的网站,wordpress 招聘网站,平面设计公司经营范围在之前的学习中#xff0c;我们只介绍了消息的发送#xff0c;但是没有考虑到异常的情况#xff0c;今天我们就介绍一些异常情况#xff0c;和细节的部分。
目录
生产者可靠性
生产者重连
生产者确认
MQ可靠性
持久化
Lazy Queue
消费者可靠性
消费者确认机制
失…在之前的学习中我们只介绍了消息的发送但是没有考虑到异常的情况今天我们就介绍一些异常情况和细节的部分。
目录
生产者可靠性
生产者重连
生产者确认
MQ可靠性
持久化
Lazy Queue
消费者可靠性
消费者确认机制
失败重试机制
业务幂等性
延迟消息
死信交换机
延迟消息插件 生产者可靠性
生产者重连
有时候由于网络波动可能会出现客户端连接MQ失败的情况。通过配置我们可以开启连接失败后的重连机制
spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数下次等待时长initial-interval * multipliermax-attempts: 3 # 最大重试次数
在停止mq服务之后运行代码会发现测试失败因为连接失败。最大重试次数设置的是3此处就重试了3次再停止。 当网络不稳定的时候利用重试机制可以有效提高消息发送的成功率。不过SpringAMOP提供的重试机制是阻塞式的重试也就是说多次重试等待的过程中当前线程是被阻塞的 会影响业务性能。 如果对于业务性能有要求建议禁用重试机制。如果一定要使用请合理配置等待时长和重试次数当然也可以考虑使用异步线程来执行发送消息的代码。 生产者确认
在开启了生产者确认机制后在MQ成功收到消息后会返回确认消息ACK给生产者如果有异常会返回NACK。
消息投递到了MQ但是路由失败。此时会通过PublisherReturn返回路由异常原因然后返回ACK告知投递成功临时消息投递到了MQ并且入队成功返回ACK告知投递成功持久消息投递到了MQ并且入队完成持久化返回ACK告知投递成功其它情况都会返回NACK告知投递失败 spring:rabbitmq:publisher-confirm-type: correlated# 开启publisher confirm机制并设置confirm类型为correlatedpublisher-returns: true# 开启publisher return机制 这里publisher-confirm-type有三种模式可选
none:关闭confirm机制simple:同步阻塞等待MO的回执消息correlated:MO异步回调方式返回回执消息
但是生产者确认需要额外的网络和系统资源开销尽量不要使用。 如果一定要使用无需开启Publisher-Return机制因为一般路由失败是自己业务问题。 对于nack消息可以有限次数重试依然失败则记录异常消息。 MQ可靠性
解决了生产者可靠性还需要解决MQ的可靠性。
通过生产者发送的消息被MQ存放到内存中经过某些特殊情况或者MQ重启后这部分数据会丢失。并且内存空间是有限的当消费者故障或者处理太慢时会导致消息积压导致MQ阻塞。
持久化
为了解决这个问题MQ引入了数据持久化包括交换机持久化、队列持久化、消息持久化。
前两者只需要在创建的时候设置成Durable默认即可。 消息持久化默认是不开启的要手动开启。 当不开启磁盘持久化消息会全部存放在内存中。但是发送消息过多会占满内存。之后多出来的消息会存放到Paged out中也就是磁盘中。等待内存中的消息被处理完后会再把磁盘中的消息加载到内存中再继续处理。
当开启了磁盘持久化接收到的消息会在内存和磁盘中都存一份。此时处理的消息是从内存中处理。内存会在将要满的时候清理一次再继续完成消息处理。磁盘则会把所有消息都保存下来。
Lazy Queue
Lazy Queue惰性队列的特征如下:
接收到消息后直接存入磁盘而非内存(内存中只保留最近的消息默认2048条消费者要消费消息时才会从磁盘中读取并加载到内存支持数百万条的消息存储 并且惰性队列的性能很高比之前的几种性能都会好一些。 消费者可靠性
消费者确认机制
当不开启消费者确认机制生产者投递了一条消息不管消费者是否处理完了会马上被RabbitMQ删除当做已经处理完了。但是如果消费者出现网络波动或者其他异常情况会导致没有接收到这条消息生产者这边还会认为消费者已经接收到消息了。告知RabbitMQ自己消息处理状态。处理消息结束后应该向RabbitMQ发送一个回执
回执有三种可选值:
ack:成功处理消息RabbitMO从队列中删除该消息nack:消息处理失败RabbitMO需要再次投递消息reiect:消息处理失败并拒绝该消息RabbitMO从队列中删除该消息
SpringAMQP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处理方式有三种方式: none不处理消息确认。消息投递给消费者后立刻ack消息会立刻从MQ删除。这种方式非常不安全不建议使用。 manual手动模式。需要在业务代码中手动调用API发送ack或reject。虽然存在业务侵入但提供了更大的灵活性。 auto自动模式。SpringAMQP利用AOP对消息处理逻辑进行环绕增强。当业务正常执行时自动返回ack。当业务出现异常时根据异常类型自动返回不同的结果 1.如果是业务异常自动返回nack。 2.如果是消息处理或校验异常自动返回reject。
spring:rabbitmq:listener:simple:acknowledge-mode: auto # 可以设置为 none, manual, auto 失败重试机制
当消费者出现异常后消息会不断requeue重新入队到队列再重新发送给消费者然后再次异常再次requeue形成无限循环。这会导致MQ的消息处理量飙升给系统带来不必要的压力。
我们可以利用Spring的retry机制在消费者出现异常时进行本地重试而不是无限制地requeue到MQ队列。并且指定最大重试次数。
spring:rabbitmq:listener:simple:prefetch: 1retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初始的失败等待时长为1秒multiplier: 1 # 下次失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态; false有状态。如果业务中包含事务这里改为false 在开启重试模式后如果重试次数耗尽且消息依然失败则需要有MessageRecoverer接口来处理。MessageRecoverer包含三种不同的实现
RejectAndDontRequeueRecoverer重试耗尽后直接reject丢弃消息。默认就是这种方式。ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队。RepublishMessageRecoverer重试耗尽后将失败消息投递到指定的交换机
当通过最后一种方式重试耗尽后我们可以额外设置一个队列比如error.queue当发送失败的消息进入这个队列后再通过邮件强提醒这样的机制推送给工作人员可以有效解决消息发送失败的极端情况。 业务幂等性
在程序开发中业务幂等性指的是同一个业务执行一次或者执行多次对业务的状态是没有影响的。
唯一消息id 方案一是给每个消息都设置一个唯一id利用id区分是否是重复消息:
每一条消息都生成一个唯一的id与消息一起投递给消费者。消费者接收到消息后处理自己的业务业务处理成功后将消息ID保存到数据库如果下次又收到相同消息去数据库查询判断是否存在存在则为重复消息放弃处理。
业务判断
方案二是结合业务逻辑基于业务本身做判断。以我们的业务为例:我们要在支付后修改订单状态为已支付应该在修改订单状态前先查询订单状态判断状态是否是未支付。只有未支付订单才需要修改其它状态不做处理。 如何保证支付服务与交易服务之间的订单状态一致性 为确保支付服务与交易服务之间的订单状态一致性我们采取了以下措施 消息通知 支付服务在用户支付成功后通过MQ消息队列发送消息通知交易服务以完成订单状态的同步。 消息可靠性策略 采用生产者确认机制、消费者确认和消费者失败重试等策略确保消息的可靠投递和处理。开启MQ的持久化功能避免因服务宕机导致消息丢失。 业务幂等性 在交易服务更新订单状态时进行业务幂等性判断防止因消息重复消费导致订单状态异常。 如果交易服务消息处理失败有什么兜底方案 定时任务 在交易服务中设置定时任务定期查询订单的支付状态。即使MQ通知失败定时任务也可以作为兜底方案确保订单支付状态的最终一致性。 延迟消息
延迟消息是消息队列中的一种重要功能它允许消息在被发送到消息队列后并不会立即被消费者消费而是在经过特定的时间延迟后才能被消费者获取和处理。这种特性在很多业务场景中都非常有用比如订单处理超时、定时提醒等。
比如用户A下单某商品的最后一件订单确认后迟迟不支付但是又占用着这个名额。等到很久以后取消这个订单此时想买的人没买到商家没有卖掉而这个人又没有买。这种情况用延迟消息就能很好的解决这个问题。当用户下单商品后会设置一个延迟消息假设30分钟内没有下单这个延迟消息就会被发送到MQ提醒数据库这个人订单超时了强制让这个人取消订单。
死信交换机
当一个队列中的消息满足下列情况之一时就会成为死信(dead letter)
消费者使用basic.reject或 basic.nack声明消费失败并且消息的requeue参数设置为false消息是一个过期消息(达到了队列或消息本身设置的过期时间)超时无人消费要投递的队列消息堆积满了最早的消息可能成为死信
死信消息经过死信交换机可以变成延迟消息。 当publisher发布一个消息后通过交换机进入队列。通过手动设置一个过期时间让消息变成死信消息此时消息会自动进入通过dead-letter-exchang设置的交换机dlx.direct再一步步的进入到consumer。
延迟消息插件 RabbitMO的官方也推出了一个插件原生支持延迟消息功能。该插件的原理是设计了一种支持延迟消息功能的交换机 当消息投递到交换机后可以暂存一定时间到期后再投递到队列。
RabbitListener(bindingsQueueBinding(
valueQueue(namedelay.queue, durabletrue),
exchangeExchange(namedelay.direct,delayedtrue),
keydelay
))
public void listenDelayMessage(String msg){
log.info(接收到delay.queue的延迟消息:{}msg);
}
Bean
public DirectExchange delayExchange(){
return ExchangeBuilder
.directExchange(delay.direct)
.delayed()//设置delay的属性为true
.durable(true)//持久化
.build();
}
发送消息时需要通过消息头x-delay来设置过期时间:
Test
void testPublisherDelayMessage() {//1.创建消息String message hello, delayed message;//2.发送消息利用消息后置处理器添加消息头rabbitTemplate.convertAndSend(delay.direct, delay, message, new MessagePostProcessor() {Overridepublic Message postProcessMessage(Message message) throws AmqpException {//添加延迟消息属性message.getMessageProperties().setDelay(5000);return message;}});
}
