【RabbitMQ】可靠性策略（幂等，消息持久化）

MQ可靠性策略

发送者的可靠性问题

生产者的重连

可能存在由于网络波动，出现的客户端连接MQ失败，我们可以通过配置文件配置解决

spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled:true#开启超时重试机制initial-interval: 1000ms #失败后的初始等待时间multiplier:1#失败后下次的等待时长倍数，下次等待时间= initial-interval * multipliermax-attempts:3#最大重试次数

生产者确认

RabbitMQ提供了Publisher Confirm和Publisher Return两种确认机制，开机确认机制后，在MQ成功收到消息后会返回确认消息给生产者，返回的结果有以下几种情况：

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

通过配置文件配置生产者的消息类型：

spring:rabbitmq:publisher-confirm-type: correlated #开启publisher confirm机制，并设置confirm类型publisher-returns:true# 开启publisher return机制

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

none

:关闭confirm机制

simple

: 同步阻塞等待MQ的回执消息

correlated

:MQ异步调用方式返回回执消息

异步回调方式：

我们完成一个任务将消息交由消息队列中，就进行别的任务了，当消息队列返回异常问题，在过来进行对应的处理

我们需要调用ReturnCallback函数完成消息失败后的操作：
在使用之前需要配置ReturnCallback，每个RabbitTemplate只能配置一个ReturnCallback

@ConfigurationpublicclassCommonConfigimplementsApplicationContextAware{@OverridepublicvoidsetApplicationContext(ApplicationContext applicationContext)throwsBeansException{//获取RabbitTemplateRabbitTemplate rabbittemplate =applicationContext.getBean(RabbitTemplate.class);//设置ReturnCallback
   rabbitTemplate.setReturnCallback(message,replyCode,replyText,exchange,routingKey)->{//处理操作}}}

通过ConfirmCallback来处理消息失败:
每一个消息指定一个ConfirmCallback

voidtest()throwsInterruptedException{CorrelationData cd=newCorrelationData();
    cd.getFuture().addCallback(newListenableFutureCallback<CorrelationData.confirm>(){@OverridepublicvoidonFailure(Throwable ex){//Future 发生异常是的逻辑处理，基本不会触发}@OverridepublicvoidonSuccess(CorrelationData.confirm result){//Future接收到回执的处理逻辑，参数中的result就是回执内容if(result.isAck()){//result.isack,boolean类型，true代表ack回执，false表示nack回执//处理逻辑}else{//异常处理}}});}
rabbittemplate.convertAndSend("","",cd);

MQ的可靠性

在默认情况下，Rabbitmq会将接收到的数据保存在内存中以降低消息收发的延迟，这样会有问题：

1. 一旦MQ宕机，内存的消息会丢失
2. 内存空间有限，当消费者故障或处理过慢时，会导致消息积压，引发MQ阻塞（消息对队列将消息保存到磁盘，此时MQ阻塞）

数据持久化

RabbitMQ的数据持久化包括：

1. 交换机持久化（Durable 永久的，Transient临时的）
2. 队列持久化（Durable 永久的，Transient临时的）
3. 消息持久化

消息的持久化：

voidtest(){Message message =MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF-8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();}

会将消息在过程中持久化到磁盘不会导致MQ阻塞

消息的持久化性能不是很高，可以通过Lazy Queue进行消息的持久化

Lazy Queue

惰性队列
特征：

1. 接收到消息后直接存入磁盘而非内存（内存只保留最近消息，默认2048条）
2. 消费者要消费消息时才会从磁盘中读取并加载到内存
3. 支持百万条数据的消息存储

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

在java中要设置一个队列为惰性队列，只需要在声明队列时，指定

x-queue-mode

属性为lazy：

@BeanpublicQueuelazyQueue(){returnQueueBuilder.durable("lazy.queue").lazy()//开启lazy模式.build();}

通过注解也可以实现

@RabbitListener(queuesToDeclare=@Queue(
                name="lazy.queue",
                durable="true",//持久化
                arguments =@Argument(name="x-queue-mode",value="lazy")))publicvoidlistenLazyQueue(String msg){//消费处理}

消费者的可靠性问题

消费者确认机制

为了确认消费者是否成功处理消息，RabbitMQ提供了消费者确认机制，当消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态。回执有三种可选值:

ack:

成功处理消息，RabbitMQ从队列中删除该信息

nack：

消息处理失败，RabbitMQ需要再次投递信息

reject：

消息处理失败并拒绝该消息，RabbitMQ从队列中删除该信息
那么如何实现呢：
SprinaAMOP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处理方式
有三种方式：

none:

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

manual:

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

auto:

自动模式。SprinGAMOP利用AOP对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回ack.当业务出现异常时：
如果是业务异常，则会自动返回nack
如果是消息处理或校验异常，自动返回reject

spring:rabbitmq:listener:simple:prefetch:1acknowledge-mode: none #none，关闭

消息失败处理

如果消费者返回nack，那就会重复进行，这样大大影响效率
我们可以利用Spring的retry机制，在消费者出现异常的时利用本地重试：

spring:rabbitmq:listener:simple:prefetch:1retry:enabled:true#开启消费者失败重试initial-interval: 1000ms #初始的失败等待时间multiplier:1#下次失败的等待时长的倍数max-attempts:3# 最大重试次数stateless:true# true无状态，false有状态，如果业务中包含事务，这里改为false

在开启重试模式之后，重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口来处理，它包含三种实现：

RejectAndDontRequeueRecoverer

:重试耗尽后，直接reject，丢弃消息，默认这种方式

ImmediateRequeueMessageRecoverer

：重试耗尽后，返回nack，消息重新入队

RepublishMessageRecoverer

:重试耗尽后，将失败消息投递到指定的交换机

将失败消息重新投递到error交换机中，可以绑定error消息队列，将来发送信息给开发人员等操作消息

将失败策略改为RepublishMessageRecoverer：

1. 首先，定义接收失败消息的交换机，队列绑定
2. 定义RepublishMessageRecoverer

@BeanpublicMessageRecoverertest(RabbitTemplate rabbittemplate){returnnewRepublishMessageRecoverer(rabbitTemplate,"交换机名称","key值")}

业务幂等性

幂等是一个数学概念，用函数表达式来描述是这样的：f(x)=f(f(x)),在程序开发中，则指同一个业务，执行一次或多次对业务状态的影响是一致的

幂等的使用场景

：防止某一数据被重复进行修改

幂等业务

：根据id的查询业务，根据id的删除业务等

非幂等

：用户下单，扣减库存等

如何实现幂等：
方案一唯一消息id
给每一个消息都设置一个唯一id，利用id区分是否重复消息：

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

@BeanpublicMessageConvertermessageConverter(){//定义消息转换器Jackson2JsonMessageConverter jjmc=newjackson2JsonMessageConverter();//配置自动创建消息id，用于识别不同消息，也可以在业务中基于id判断是否是重复消息
    jjmc.setCreateMessageIds(true);return jjmc;}

但是这样会造成额外的操作冗余，比如还需要写数据库等等
方案二：结合业务逻辑，基于业务本身做判断

简答问题

如何保证支付服务与交易服务之间的订单状态一致性：

1. 首先，支付服务会在正在用户支付成功以后利用MQ消息通知交易服务，完成订单状态同步
2. 其次，为了保证消息的可靠性，我们采取了生产者确认机制，消费者确认，消费者失败重试等策略，确保消息投递和处理的可靠性，同时可开启了MQ的持久化，避免因服务宕机导致消息丢失
3. 最后，我们还会交易服务更新订单状态时作业业务幂等判断，避免因消息重复导致订单异常

如果交易服务处理失败，还有什么方案：
在交易服务设置定时任务，定期查询订单生产状态，这样即使MQ通知失败，还可以利用定时任务作为兜底方案，确保订单支付状态的最终一致性