RabbitMQ消费者的可靠性

目录

一、消费者确认

二、失败重试机制

2.1、失败处理策略

三、业务幂等性

3.1、唯一消息ID

 3.2、业务判断

3.3、兜底方案


一、消费者确认

RabbitMQ提供了消费者确认机制(Consumer Acknowledgement)。即:当消费者处理消息结束后,应该向RabbitMQ发送一个回执,告知RabbitMQ自己消息处理状态。回执有三种可选值:

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

一般reject方式用的较少,除非是消息格式有问题,那就是开发问题了。因此大多数情况下我们需要将消息处理的代码通过try catch机制捕获,消息处理成功时返回ack,处理失败时返回nack.

由于消息回执的处理代码比较统一,因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式,有三种模式:

  • none:不处理。即消息投递给消费者后立刻ack,消息会立刻从MQ删除。非常不安全,不建议使用
  • manual:手动模式。需要自己在业务代码中调用api,发送ackreject,存在业务入侵,但更灵活
  • auto:自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强,当业务正常执行时则自动返回ack. 当业务出现异常时,根据异常判断返回不同结果:
    • 如果是业务异常,会自动返回nack
    • 如果是消息处理或校验异常,自动返回reject;

 通过下面的配置可以修改SpringAMQP的ACK处理方式:

spring:rabbitmq:host: 192.168.200.129 # 你的虚拟机IPport: 5672 # 端口virtual-host: / # 虚拟主机username: admin # 用户名password: 123 # 密码listener:simple:prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息acknowledge-mode: none

当使用none模式时

生产者发送一条消息

 消费者接受消息时抛异常

 

测试可以发现:当消息处理发生异常时,消息依然被RabbitMQ删除了。

把确认机制修改为auto

spring:rabbitmq:host: 192.168.200.129 # 你的虚拟机IPport: 5672 # 端口virtual-host: / # 虚拟主机username: admin # 用户名password: 123 # 密码listener:simple:prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息acknowledge-mode: auto

 再次发送消息

 

在异常位置打断点,再次发送消息,程序卡在断点时,可以发现此时消息状态为unacked(未确定状态):

当我们把配置改为auto时,消息处理失败后,会回到RabbitMQ,并重新投递到消费者。

二、失败重试机制

当消费者出现异常后,消息会不断requeue(重入队)到队列,再重新发送给消费者。如果消费者再次执行依然出错,消息会再次requeue到队列,再次投递,直到消息处理成功为止。 极端情况就是消费者一直无法执行成功,那么消息requeue就会无限循环,导致mq的消息处理飙升,带来不必要的压力。

为了应对上述情况Spring又提供了消费者失败重试机制:在消费者出现异常时利用本地重试,而不是无限制的requeue到mq队列。

修改consumer服务的application.yml文件,添加内容

spring:rabbitmq:host: 192.168.200.129 # 你的虚拟机IPport: 5672 # 端口virtual-host: / # 虚拟主机username: admin # 用户名password: 123 # 密码listener:simple:prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息acknowledge-mode: auto #消息确认retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数,下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态;false有状态。如果业务中包含事务,这里改为false

重启consumer服务,重复之前的测试。可以发现:

  • 消费者在失败后消息没有重新回到MQ无限重新投递,而是在本地重试了3次

结论:

  • 开启本地重试时,消息处理过程中抛出异常,不会requeue到队列,而是在消费者本地重试
  • 重试达到最大次数后,Spring会返回reject,消息会被丢弃
2.1、失败处理策略

在之前的测试中,本地测试达到最大重试次数后,消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下,显然不太合适了

因此Spring允许我们自定义重试次数耗尽后的消息处理策略,这个策略是由MessageRecovery接口来定义的,它有3个不同实现:

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

在消费者里创建一个异常消息配置类

@Configuration
@Slf4j
//当配置文件中spring.rabbitmq.listener.simple.retry.enabled 属性为ture时配置类才生效
@ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
public class ErrorMessageConfig {//消息处理失败交换机@Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");}//消息处理失败队列@Beanpublic Queue errorQueue(){return new Queue("error.queue", true);}//绑定关系@Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");}//定义一个RepublishMessageRecoverer,关联队列和交换机@Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){log.error("加载RepublishMessageRecoverer");return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");}
}

 当接收消息出现异常时,会创建error.queue队列

 可以查看到异常信息

三、业务幂等性

何为幂等性? 幂等是一个数学概念,用函数表达式来描述是这样的:f(x) = f(f(x)),例如求绝对值函数。 在程序开发中,则是指同一个业务,执行一次或多次对业务状态的影响是一致的。例如:

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

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

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

所以,我们要尽可能避免业务被重复执行。 然而在实际业务场景中,由于意外经常会出现业务被重复执行的情况,例如:

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

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

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

因此,我们必须想办法保证消息处理的幂等性。这里给出两种方案:

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

SpringAMQP的MessageConverter自带了MessageID的功能,我们只要开启这个功能即可。 以Jackson的消息转换器为例 

在生产者和消费者的启动类里加一个配置

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

 测试生产者发送消息会产生一个id

 3.2、业务判断

业务判断就是基于业务本身的逻辑或状态来判断是否是重复的请求或消息,不同的业务场景判断的思路也不一样。 例如我们当前案例中,处理消息的业务逻辑是把订单状态从未支付修改为已支付。因此我们就可以在执行业务时判断订单状态是否是未支付,如果不是则证明订单已经被处理过,无需重复处理。

相比较而言,消息ID的方案需要改造原有的数据库,所以我更推荐使用业务判断的方案。

以支付修改订单的业务为例,我们需要修改OrderServiceImpl中的markOrderPaySuccess方法:

 @Overridepublic void markOrderPaySuccess(Long orderId) {// 1.查询订单Order old = getById(orderId);// 2.判断订单状态if (old == null || old.getStatus() != 1) {// 订单不存在或者订单状态不是1,放弃处理return;}// 3.尝试更新订单Order order = new Order();order.setId(orderId);order.setStatus(2);order.setPayTime(LocalDateTime.now());updateById(order);}

上述代码逻辑上符合了幂等判断的需求,但是由于判断和更新是两步动作,因此在极小概率下可能存在线程安全问题。

@Override
public void markOrderPaySuccess(Long orderId) {// UPDATE `order` SET status = ? , pay_time = ? WHERE id = ? AND status = 1lambdaUpdate().set(Order::getStatus, 2).set(Order::getPayTime, LocalDateTime.now()).eq(Order::getId, orderId).eq(Order::getStatus, 1).update();
}

注意看,上述代码等同于这样的SQL语句:

UPDATE `order` SET status = ? , pay_time = ? WHERE id = ? AND status = 1

我们在where条件中除了判断id以外,还加上了status必须为1的条件。如果条件不符(说明订单已支付),则SQL匹配不到数据,根本不会执行。

3.3、兜底方案

我们可以在交易服务设置定时任务,定期查询订单支付状态。这样即便MQ通知失败,还可以利用定时任务作为兜底方案,确保订单支付状态的最终一致性。
 

 

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/124591.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

力扣刷题 day60:10-30

1.Excel表列名称 给你一个整数 columnNumber ,返回它在 Excel 表中相对应的列名称。 例如: A -> 1 B -> 2 C -> 3 ... Z -> 26 AA -> 27 AB -> 28 ... 方法一 :进制转换 #方法一:进制转换 def convertToTi…

JDK项目分析的经验分享

基本类型的包装类(Character放在最后) String、StringBuffer、StringBuilder、StringJoiner、StringTokenizer(补充正则表达式的知识) CharacterIterator、StringCharacterIterator、CharsetProvider、CharsetEncoder、CharsetDecoder(较难) java.util.function下的函数表…

网络新闻发稿为何经久不衰?

有的老板可能看不到新闻营销的直接回报,一直不乐意在此方面投入,但是却看到竞争对手一直在搞新闻营销,也就安排个PR做做新闻公关。小马识途营销顾问观察,自互联网诞生以来,新闻营销一直是网络营销工作中的一个重点。 如…

C/C++选择排序算法

文章目录 例题原理阐述代码实现总结 例题 原理阐述 从小到大排序,每次选择序列的最小的一个数; 将原来的序列A分割为排好序的序列a和未排序的序列b; 从未排好序的序列b中找到最小值,插入到排好序的序列a后面,形成从小…

【Linux】安装配置解决CentosMobaXterm的使用及Linux常用命令以及命令模式

目录 Centos的介绍 centos安装配置&MobaXterm 创建 安装 ​编辑 配置 ​编辑 MobaXterm使用 Linux常用命令&模式 常用命令 vi或vim编辑器 三种模式 命令模式 编辑模式 末行模式 拍照备份 Centos的介绍 CentOS(Community Enterprise Op…

用大白话聊聊SpringBoot的自动配置原理(面试题详解)

首先,SpringBoot的自动配置不等于自动装配! 自动配置是Auto-Configuration,针对的是SpringBoot中的配置类, 而自动装配是Autowire,针对的是Spring中的依赖注入。 进入主题: 自动配置简单来说就是自动去把…

在Qt中List View和List Widget的区别是什么,以及如何使用它们

2023年10月29日,周日晚上 目录 List View和List Widget的区别 如何使用QListView 如何使用QListWidget List View和List Widget的区别 在Qt中,QListView 和 QListWidget 是用于显示列表数据的两个常用控件,它们有一些区别和特点。 1. 数…

Android应用:实现网络加载商品数据【OKHttp、Glide、Gson】

实现网络加载商品数据的功能: 1、在AndroidManifest.xml中声明网络权限; 2、在app/build.gradle中添加okhttp, glide, gson等必需的第3方库; 3、在MainActivity中通过OkHttpClient连接给定的Web服务,获取商品数据;对…

力扣每日一题86:分隔链表

题目描述: 给你一个链表的头节点 head 和一个特定值 x ,请你对链表进行分隔,使得所有 小于 x 的节点都出现在 大于或等于 x 的节点之前。 你应当 保留 两个分区中每个节点的初始相对位置。 示例 1: 输入:head [1,4,…

开源库存管理系统InvenTree的安装

本文是应网友 shijie880500 要求折腾的; 什么是 InvenTree ? InvenTree 是一个开源的库存管理系统,提供强大的低级别库存控制和零件跟踪。InvenTree 系统的核心是 Python/Django 数据库后端,它提供了一个管理界面(基于…

CNN 网络结构简介

本文通过整理李宏毅老师的机器学习教程的内容,介绍 CNN(卷积神经网络)的网络结构。 CNN 网络结构, 李宏毅 CNN 主要应用在图像识别(image classification, 图像分类)领域。 通常,输入的图片大小相同&am…

案例分析真题-系统建模

案例分析真题-系统建模 2009年真题 【问题1】 【问题2】 【问题3】 2012年真题 【问题1】 【问题2】 【问题3】 2014年真题 【问题1】 【问题2】 骚戴理解:这个题目以前经常考,不知道今年会不会考,判断的话就是看加工有没有缺少输入和输出&a…

基于nodejs+vue 学生考勤综合平台的设计与实现

目 录 摘 要 I ABSTRACT II 目 录 II 第1章 绪论 1 1.1背景及意义 1 1.2 国内外研究概况 1 1.3 研究的内容 1 第2章 相关技术 3 2.1 nodejs简介 4 2.2 express框架介绍 6 2.4 MySQL数据库 4 第3章 系统分析 5 3.1 需求分析 5 3.2 系统可行性分析 5 3.2.1技术可行性:…

AMD:抢占AI芯片宝座

来源:猛兽财经 作者:猛兽财经 总结: (1)AMD受益于AI芯片的出口限制,使其能够获得更多的中国市场份额,并增强其在AI芯片市场的地位。 (2)AMD的处理器,特别是E…

springboot动态数据源【非伪数据源】

说明&#xff1a;本文章的数据源不是在配置文件中配置两个或多个数据源&#xff0c;在业务方面对这些数据源来回切换&#xff0c;本文章中的数据源是可以动态添加&#xff0c;修改&#xff0c;切换的&#xff0c;废话不多说。 先看工程图&#xff1a; 1.pom.xml文件 <?x…

多输入多输出 | Matlab实现k-means-ELM(k均值聚类结合极限学习机)多输入多输出组合预测

多输入多输出 | Matlab实现k-means-ELM&#xff08;k均值聚类结合极限学习机&#xff09;多输入多输出组合预测 目录 多输入多输出 | Matlab实现k-means-ELM&#xff08;k均值聚类结合极限学习机&#xff09;多输入多输出组合预测预测效果基本描述程序设计参考资料 预测效果 基…

nodejs+python+php+微信小程序-基于安卓android的健身服务应用APP-计算机毕业设计

考虑到实际生活中在健身服务应用方面的需要以及对该系统认真的分析&#xff0c;将系统权限按管理员和用户这两类涉及用户划分。  则对于进一步提高健身服务应用发展&#xff0c;丰富健身服务应用经验能起到不少的促进作用。 健身服务应用APP能够通过互联网得到广泛的、全面的宣…

ES 8.x新特性一览(完整版)

一、看点 在 2022 年 2 月 11 日&#xff0c;Elasticsearch&#xff08;ES&#xff09;正式发布了 8.0 版本&#xff0c;而截止到 2023 年 10 月&#xff0c;历经一年半时间&#xff0c;ES官方已经连续发布了多个版本&#xff0c;最新版本为 8.10.4。这一系列的更新引入了众多引…

小红书app拉新上线了 适合网推社群和校园渠道作业

小红书app签到拉新上线了可以通过“聚量推客”进行申请&#xff0c;下面大概是要求和流程 要求网推社群渠道或者地推校园渠道&#xff0c;其它类型渠道禁止

sql--索引使用 ---覆盖索引

覆盖索引 Select 后接 * 走id索引才是最优&#xff0c;使用二级索引则需要回表&#xff08;性能稍差&#xff09; 前缀索引 Create index 索引名 on 表名( 字段名( n ) ) n数字 n代表提取这个字符串的n个构建索引 &#xff1f;&#xff1f;那么 n 为几性能是最好的呢&…