0


【RabbitMQ】-消息可靠性以及延迟消息

发送者的可靠性

发送者重连

有的时候由于网络波动,可能会出现发送者连接MQ失败的情况。通过配置我们可以开启连接失败后的重连机制:

【RabbitMQ】-消息可靠性以及延迟消息_持久化

注意:当网络,稳定的时候,利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试,也就是说多次重试等待的过程中,当前线程是被阻塞的,会影响业务性能。

如果对于业务性能有要求,建议禁用重试机制。如果一定要使用,请合理配置等待时长和重试次数,当然也可以考虑使用异步线程来执行发送消息的代码。

发送者确认

SpringAMQP提供了Publisher Confirm和Publisher Return两种确认机制。开启确机制认后,当发送者发送消息给MQ后,MQ会返回确认结果给发送者。返回的结果有以下几种情况:

  • 消息投递到了MQ,但是路由失败。此时会通过PublisherReturn返回路由异常原因,然后返回ACK,告知投递成功。
  • 临时消息投递到了MQ,并且入队成功,返回ACK,告知投递成功
  • 持久消息投递到了MQ,并且入队完成持久化,返回ACK,告知投递成功
  • 其它情况都会返回NACK,告知投递失败

【RabbitMQ】-消息可靠性以及延迟消息_持久化_02

SpringAMQP实现发送者确认

1)在publisher这个微服务的app;ication.yml中添加配置:

【RabbitMQ】-消息可靠性以及延迟消息_消息处理_03

配置说明:

这里publisher-confirm-type有三种模式可选:

  • none:关闭confirm机制
  • simple:同步阻塞等待MQ的回执消息
  • correlated:MQ异步回调方式返回回执消息

2)每个RabbitTemplate只能配置一个Returncallback,因此需要在项目启动过程中配置:

【RabbitMQ】-消息可靠性以及延迟消息_消息处理_04

3)发送消息,指定消息ID、消息ConfirmCallback

【RabbitMQ】-消息可靠性以及延迟消息_发送消息_05

MQ的可靠性

在默认情况下,RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。这样会导致两个问题:

一旦MQ宕机,内存中的消息会丢失

内存空间有限,当消费者故障或处理过慢时,会导致消息积压,引发MQ阻塞

数据持久化

RabbitMQ实现数据持久化包括3个方面:

  • 交换机持久化
  • 队列持久化
  • 消息持久化

Lazy Queue

在默认情况下,RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。但在某些特殊情况下,这会导致消息积压,比如:

  • 消费者宕机或出现网络故障
  • 消息发送量激增,超过了消费者处理速度
  • 消费者处理业务发生阻塞

一旦出现消息堆积问题,RabbitMQ的内存占用就会越来越高,直到触发内存预警上限。此时RabbitMQ会将内存消息刷到磁盘上,这个行为成为

PageOut

.

PageOut

会耗费一段时间,并且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处理新的消息,生产者的所有请求都会被阻塞。

从RabbitMQ的3.6.0版本开始,就增加了Lazy Queue的概念,也就是惰性队列。惰性队列的特征如下:

  • 接收到消息后直接存入磁盘,不再存储到内存
  • 消费者要消费消息时才会从磁盘中读取并加载到内存(可以提前缓存部分消息到内存,最多2048

在3.12版本后,所有队列都是Lazy Queue模式,无法更改。

要设置一个队列为惰性队列,只需要再声明队列时,指定x-queue-mode属性为lazy即可:

【RabbitMQ】-消息可靠性以及延迟消息_发送消息_06

【RabbitMQ】-消息可靠性以及延迟消息_发送消息_07

或者

【RabbitMQ】-消息可靠性以及延迟消息_发送消息_08

RabbitMQ如何保证消息的可靠性

首先通过配置可以让交换机、队列、以及发送的消息都持久化。

这样队列中的消息会持久化到磁盘,MQ重启消息依然存在。

RabbitMQ在3.6版本引入了LazyQueue,并且在3.12版本后会称为队列的默认模式。LazyQueue会将所有消息都持久化。

开启持久化和生产者确认时,RabbitMQ只有在消息持久化完成后才会给生产者返回ACK回执

消费者的可靠性

消费者确认机制

消费者确认机制(Consumer Acknowledgement)是为了确认消费者是否成功处理消息。当消费者处理消息结束后,应该向RabbitMQ发送一个回执,告知RabbitMQ自己消息处理状态:

  • ack:成功处理消息,RabbitMQ从队列中删除该消息
  • nack:消息处理失败,RabbitMQ需要再次投递消息
  • reject:消息处理失败并拒绝该消息,RabbitMQ从队列中删除该消息

【RabbitMQ】-消息可靠性以及延迟消息_发送消息_09

SpringAMQP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处理方式,有三种方式:

none:不处理。即消息投递给消费者后立刻ack,消息会立刻从MQ删除。非常不安全,不建议使用

manual:手动模式。需要自己在业务代码中调用api,发送ack或reject,存在业务入侵,但更灵活

auto:自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强,当业务正常执行时则自动返回ack.当业务出现异常时,根据异常判断返回不同结果:

  • 如果是业务异常,会自动返回nack
  • 如果是消息处理或校验异常,自动返回reject

【RabbitMQ】-消息可靠性以及延迟消息_持久化_10

失败重试机制

SpringAMQP提供了消费者失败重试机制,在消费者出现异常时利用本地重试,而不是无限的requeue到mq。我们可以通过在application.yaml文件中添加配置来开启重试机制:

【RabbitMQ】-消息可靠性以及延迟消息_持久化_11

失败消息处理策略

在开启重试模式后,重试次数耗尽,如果消息依然失败,则需要有NessageRecoverer接口来处理,它包含三种不同的实现:

  • RejectAndDontRequeueRecoverer:重试耗尽后,直接reject,丢弃消息。默认就是这种方式
  • ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队
  • RepublishMessageRecoverer:重试耗尽后,将失败消息投递到指定的交换机

【RabbitMQ】-消息可靠性以及延迟消息_消息处理_12

将失败处理策略改为RepublishMessageRecoverer:

首先,定义接收失败消息的交换机、队列及其绑定关系

然后,定义RepublishMessageRecoverer:

【RabbitMQ】-消息可靠性以及延迟消息_消息处理_13

登录后复制

publicclassErrorMessageConfiguration{@BeanpublicDirectExchangeerrorExchange(){returnnewDirectExchange("error.direct");}@BeanpublicQueueerrorQueue(){returnnewQueue("error.queue");}@BeanpublicBindingerrorQueueBinding(Queue errorQueue,DirectExchange errorExchange){returnBindingBuilder.bind(errorQueue).to(errorExchange).with("error");}@BeanpublicMessageRecoverermessageRecoverer(RabbitTemplate rabbitTemplate){returnnewRepublishMessageRecoverer(rabbitTemplate,"error.direct","error");}}

业务幂等性

何为幂等性?

幂等是一个数学概念,用函数表达式来描述是这样的:

f(x) = f(f(x))

,例如求绝对值函数。

在程序开发中,则是指同一个业务,执行一次或多次对业务状态的影响是一致的。例如:

  • 根据id删除数据
  • 查询数据
  • 新增数据

但数据的更新往往不是幂等的,如果重复执行可能造成不一样的后果。比如:

  • 取消订单,恢复库存的业务。如果多次恢复就会出现库存重复增加的情况
  • 退款业务。重复退款对商家而言会有经济损失。

所以,我们要尽可能避免业务被重复执行。

然而在实际业务场景中,由于意外经常会出现业务被重复执行的情况,例如:

  • 页面卡顿时频繁刷新导致表单重复提交
  • 服务间调用的重试
  • MQ消息的重复投递

我们在用户支付成功后会发送MQ消息到交易服务,修改订单状态为已支付,就可能出现消息重复投递的情况。如果消费者不做判断,很有可能导致消息被消费多次,出现业务故障。

举例:

  1. 假如用户刚刚支付完成,并且投递消息到交易服务,交易服务更改订单为已支付状态。
  2. 由于某种原因,例如网络故障导致生产者没有得到确认,隔了一段时间后重新投递给交易服务。
  3. 但是,在新投递的消息被消费之前,用户选择了退款,将订单状态改为了已退款状态。
  4. 退款完成后,新投递的消息才被消费,那么订单状态会被再次改为已支付。业务异常。

【RabbitMQ】-消息可靠性以及延迟消息_消息处理_14

处理方案

唯一消息Id

方案一,是给每个消息都设置一个唯一id,利用id区分是否是重复消息:

  • 每一条消息都生成一个唯一的id,与消息一起投递给消费者。
  • 消费者接收到消息后处理自己的业务,业务处理成功后将消息ID保存到数据库
  • 如果下次又收到相同消息,去数据库查询判断是否存在,存在则为重复消息放弃处理。

【RabbitMQ】-消息可靠性以及延迟消息_消息处理_15

登录后复制

@BeanpublicMessageConvertermessageConverter(){Jackson2JsonMessageConverter jjmc =newJackson2JsonMessageConverter();
        jjmc.setCreateMessageIds(true);return jjmc;}
业务判断

结合业务逻辑,基于业务本身做判断

【RabbitMQ】-消息可靠性以及延迟消息_持久化_16

登录后复制

publicvoidlistenPaySuccess(Long orderId){//1.查询订单Order order = orderService.getById(orderId);//2.判断订单是否为支付if(order ==null|| order.getStatus()!=1){return;}
        orderService.markOrderPaySuccess(orderId);}

延迟消息

延迟消息:发送者发送消息时指定一个时间,消费者不会立刻收到消息,而是在指定时间之后才收到消息。

延迟任务:设置在一定时间之后才执行的任务

【RabbitMQ】-消息可靠性以及延迟消息_发送消息_17

死信交换机

当一个队列中的消息满足下列情况之一时,就会成为死信(dead letter) :

  • 消费者使用basic.reject或 basic.nack声明消费失败,并且消息的requeue参数设置为false
  • 消息是一个过期消息(达到了队列或消息本身设置的过期时间),超时无人消费
  • 要投递的队列消息堆积满了,最早的消息可能成为死信

如果队列通过dead-letter-exchange属性指定了一个交换机,那么该队列中的死信就会投递到这个交换机中。这个交换机称为死信交换机( Dead Letter Exchange,简称DLX)

【RabbitMQ】-消息可靠性以及延迟消息_发送消息_18

登录后复制

@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="dlx.queue1",durable ="true"),
            exchange =@Exchange(name ="dlx.direct",type =ExchangeTypes.DIRECT),
            key ={"hi"}))publicvoidlistenDlxQueue(String message){
        log.info("消费者监听到dlx.queue1的消息:【{}】",message);}

登录后复制

publicclassNormalConfiguration{@BeanpublicDirectExchangenormalExchange(){returnnewDirectExchange("normal.direct");}@BeanpublicQueuenormalQueue(){returnQueueBuilder.durable("normal.queue").deadLetterExchange("dlx.direct").build();}@BeanpublicBindingnormalExchangeBinding(Queue normalQueue,DirectExchange normalExchange){returnBindingBuilder.bind(normalQueue).to(normalExchange).with("hi");}}

发送消息并设置时间

登录后复制

@TestvoidtestSendDelayMessage(){
        rabbitTemplate.convertAndSend("normal.direct","hi","hello",newMessagePostProcessor(){@OverridepublicMessagepostProcessMessage(Message message)throwsAmqpException{
                message.getMessageProperties().setExpiration("10000");return message;}});}

延迟消息插件

这个插件可以将普通交换机改造为支持延迟消息功能的交换机,当消息投递到交换机后可以暂存一定时间,到期后再投递到队列。

【RabbitMQ】-消息可靠性以及延迟消息_持久化_19

【RabbitMQ】-消息可靠性以及延迟消息_发送消息_20

【RabbitMQ】-消息可靠性以及延迟消息_消息处理_21

发送消息时需要通过消息头x-delay来设置过期时间:

【RabbitMQ】-消息可靠性以及延迟消息_持久化_22

登录后复制

@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="delay.queue",durable ="true"),
            exchange =@Exchange(name ="delay.direct",delayed ="true"),
            key ={"hi"}))publicvoidlistenDelayQueue(String message){
        log.info("消费者监听到delay.queue的消息:【{}】",message);}

登录后复制

@TestvoidtestSendDelayMessagePlugin(){
        rabbitTemplate.convertAndSend("delay.direct","hi","hello",newMessagePostProcessor(){@OverridepublicMessagepostProcessMessage(Message message)throwsAmqpException{
                message.getMessageProperties().setDelay(10000);return message;}});}

【RabbitMQ】-消息可靠性以及延迟消息_发送消息_23

【RabbitMQ】-消息可靠性以及延迟消息_持久化_24


本文转载自: https://blog.csdn.net/cnzzs/article/details/140562936
版权归原作者 egzosn 所有, 如有侵权,请联系我们删除。

“【RabbitMQ】-消息可靠性以及延迟消息”的评论:

还没有评论