【RabbitMQ】可靠性策略(幂等,消息持久化)

embedded/2024/9/22 13:46:44/

MQ可靠性策略

  • 发送者的可靠性问题
    • 生产者的重连
    • 生产者确认
  • MQ的可靠性
    • 数据持久化
    • Lazy Queue
  • 消费者的可靠性问题
    • 消费者确认机制
    • 消息失败处理
  • 业务幂等性
  • 简答问题

发送者的可靠性问题

生产者的重连

可能存在由于网络波动,出现的客户端连接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

@Configuration
public class CommonConfig implements ApplicationContextAware{@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException{//获取RabbitTemplateRabbitTemplate rabbittemplate =applicationContext.getBean(RabbitTemplate.class);//设置ReturnCallbackrabbitTemplate.setReturnCallback(message,replyCode,replyText,exchange,routingKey)->{//处理操作}}
}

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

void test() throws InterruptedException{CorrelationData cd= new CorrelationData();cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.confirm>(){@Overridepublic void onFailure(Throwable ex){//Future 发生异常是的逻辑处理,基本不会触发}@Overridepublic void onSuccess(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. 消息持久化

消息的持久化:

void test(){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:

@Bean
public Queue lazyQueue(){return QueueBuilder.durable("lazy.queue").lazy() //开启lazy模式.build();
}

通过注解也可以实现

@RabbitListener(queuesToDeclare= @Queue(name="lazy.queue",durable="true",//持久化arguments =@Argument(name="x-queue-mode",value="lazy")))
public void listenLazyQueue(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
@Bean
public MessageRecoverer test(RabbitTemplate rabbittemplate){return new RepublishMessageRecoverer(rabbitTemplate,"交换机名称","key值")
}

业务幂等性

幂等是一个数学概念,用函数表达式来描述是这样的:f(x)=f(f(x)),在程序开发中,则指同一个业务,执行一次或多次对业务状态的影响是一致的
幂等的使用场景:防止某一数据被重复进行修改
幂等业务:根据id的查询业务,根据id的删除业务等
非幂等:用户下单,扣减库存等

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

  1. 每一条消息都生成一个唯一id,与消息一起投递给消费者
  2. 消费者接收到消息后处理自己的业务,业务处理成功后将消息id保存到数据库
  3. 如果下次又收到相同的消息,去数据库查询判断是否存在,存在则为重复消息放弃处理
@Bean
public MessageConverter messageConverter(){//定义消息转换器Jackson2JsonMessageConverter jjmc= new jackson2JsonMessageConverter();//配置自动创建消息id,用于识别不同消息,也可以在业务中基于id判断是否是重复消息jjmc.setCreateMessageIds(true);return jjmc;
}

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

简答问题

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

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

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


http://www.ppmy.cn/embedded/30919.html

相关文章

cmd查看局域网内所有设备ip

说明&#xff1a;最近碰到一个新问题&#xff0c;就是有一个安卓设备&#xff0c;安装了一个app导致死机了&#xff0c;app设置了开机重启&#xff0c;所以&#xff0c;无论重启还是关机&#xff0c;都是进来就白屏&#xff0c; 这可把人愁坏了&#xff0c;直接死循环了 无论…

头歌:RDD的创建 -Scala

第1关&#xff1a;集合并行化创建RDD 任务描述 本关任务&#xff1a;计算并输出各个学生的总成绩。 相关知识 为了完成本关任务&#xff0c;你需要掌握&#xff1a;1.集合并行化创建RDD&#xff0c;2.reduceByKey算子、foreach算子 集合创建RDD Spark会将集合中的数据拷贝到集…

Java中的Lambda表达式

Lambda表达式的标准格式 格式&#xff1a;&#xff08;形式参数&#xff09;->{代码块} 形式参数&#xff1a;如果有多个参数&#xff0c;参数之间用逗号隔开 如果没有参数&#xff0c;留空即可 ->&#xff1a;由英文中画线和大于符号组成&#xff0c;固定写法。代表着…

Mybatis之Sqlsession、Connection和Transaction三者间的关系

前言 最近在看Mybatis的源码&#xff0c;搜到这篇文章Sqlsession、Connection和Transaction原理与三者间的关系&#xff0c;debug之后发现有不少疑惑&#xff0c;于是按照原文整理了一下&#xff0c;记录下debug中的一些困惑点。 对于我们开发来讲&#xff0c;不管跟任何关系…

低代码技术在构建质量管理系统中的应用与优势

引言 在当今快节奏的商业环境中&#xff0c;高效的质量管理系统对于组织的成功至关重要。质量管理系统帮助组织确保产品或服务符合客户的期望、符合法规标准&#xff0c;并持续改进以满足不断变化的需求。与此同时&#xff0c;随着技术的不断进步&#xff0c;低代码技术作为一…

分享从零开始学习网络设备配置--任务5.2 组建旁挂式三层无线局域网

任务描述 某公司需要在原有网络中部署WLAN&#xff0c;以满足员工的移动办公需求。由于原来的有线网络较为复杂&#xff0c;为满足WLAN组网的灵活性&#xff0c;管理员小赵准备采用AC瘦AP旁挂式三层组网方案&#xff0c;AP1部署在销售部办公室&#xff0c;AP2部署在财务部办公室…

MySQL-SQL执行流程及原理

1、SQL执行流程 2、查询流程 查询缓存&#xff1a; MySQL服务器如果在查询缓存中存在该SQL语句&#xff0c;就直接将结果返回给客户端&#xff0c;没有就进入解析器解析阶段。&#xff08;MySQL 8.0 删除该功能&#xff09;解析器&#xff1a;在解析器中对SQL语句进行语法及语…

谷粒商城实战(019 RabbitMQ-发送消息)

Java项目《谷粒商城》架构师级Java项目实战&#xff0c;对标阿里P6-P7&#xff0c;全网最强 总时长 104:45:00 共408P 此文章包含第248p-第p261的内容 介绍 MQ的三大特点&#xff1a;流量削峰、应用解耦、异步处理 redis是数据缓存&#xff0c;mq是任务缓存 AMQP的MODEL 第一…