基于注解实现去重表消息防止重复消费

基于注解实现去重表消息防止重复消费

1. 背景/问题

在分布式系统中,消息队列(如RocketMQ、Kafka)的 消息重复消费 是常见问题,主要原因包括:

  • 网络抖动:生产者或消费者因网络不稳定触发消息重发。
  • 消费者超时:消费者处理时间过长,消息队列误判为失败并重新投递。
  • 集群故障转移:消费者宕机后,未完成的消息会被其他节点重新拉取。

重复消费带来的问题

  • 业务逻辑多次执行(如重复扣款、重复生成订单)。
  • 数据一致性被破坏(如库存超卖、积分累加错误)。
  • 系统资源浪费,影响性能和稳定性。

为了避免这种情况发生,需要在客户端实现一些机制来确保消息不会被重复消费,例如记录消费者已经处理的消息 ID、使用分布式锁来控制消费进程的唯一性等。这些机制能够保证消息被成功处理,同时也能够提高系统的可靠性和稳定性。

2. 什么是幂等性

幂等性 是指对同一操作的多次执行所产生的影响与一次执行的影响相同。

  • 消息消费场景:无论消息被消费多少次,最终结果应与消费一次一致。
  • 实现目标:通过幂等设计,确保业务逻辑的重复执行不会产生副作用。

3. 幂等设计

核心思路

  1. 幂等标识:为每条消息生成唯一标识(如业务ID + 消息ID),记录其处理状态。
  2. 状态管理:通过数据库或Redis维护幂等标识的状态(如“消费中”“已消费”)。
  3. 过期时间:防止因系统崩溃导致状态长期滞留,需设置合理的超时时间(如10分钟)。
[消费者接收消息]  │  ▼  
[解析消息,生成唯一幂等标识]  │  ▼  
[查询幂等标识状态]  │  
┌───────┴───────┐  
│ 存在且已消费  │           [返回成功,丢弃消息]  
└───────┬───────┘  │  
┌───────┴───────┐  
│ 存在且消费中  │           [延迟消费,等待重试]  
└───────┬───────┘  │  
┌───────┴───────┐  
│   不存在      │  
└───────┬───────┘  │  
[设置幂等标识为“消费中”,并设置过期时间]  │  ▼  
[执行业务逻辑]  │  ▼  
[业务执行成功?]  │  
┌───────┴───────┐  
│     是        │           [更新标识为“已消费”]  
│               │           [删除或保留标识]  
└───────┬───────┘  │  
┌───────┴───────┐  
│     否        │           [删除标识,允许重试]  
└───────┬───────┘  │  ▼  
[流程结束]  

4.抽象通用幂等组件

消息防重复消费幂等组件是通用的通常会提取出来也可供其他模块/服务 使用

4.1自定义幂等注解

提供了一种通用的幂等注解,并通过 SpEL 的形式生成去重表全局唯一 Key

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface NoMQDuplicateConsume {/*** 设置防重令牌 Key 前缀*/String keyPrefix() default "";/*** 通过 SpEL 表达式生成的唯一 Key*/String key();/*** 设置防重令牌 Key 过期时间,单位秒,默认 1 小时*/long keyTimeout() default 3600L;
}

4.2. 定义幂等枚举

幂等需要设置两个状态,消费中和已消费,创建对应的枚举

@RequiredArgsConstructor
public enum IdempotentMQConsumeStatusEnum {/*** 消费中*/CONSUMING("0"),/*** 已消费*/CONSUMED("1");@Getterprivate final String code;/*** 如果消费状态等于消费中,返回失败** @param consumeStatus 消费状态* @return 是否消费失败*/public static boolean isError(String consumeStatus) {return Objects.equals(CONSUMING.code, consumeStatus);}
}

4.3.通过 AOP 的方式进行增强注解

如果说方法上加了注解,会被这段 AOP 代码以环绕增强方式执行

@Slf4j
@Aspect
@RequiredArgsConstructor
public final class NoMQDuplicateConsumeAspect {private final StringRedisTemplate stringRedisTemplate;private static final String LUA_SCRIPT = """local key = KEYS[1]local value = ARGV[1]local expire_time_ms = ARGV[2]return redis.call('SET', key, value, 'NX', 'GET', 'PX', expire_time_ms)""";/*** 增强方法标记 {@link NoMQDuplicateConsume} 注解逻辑*/@Around("@annotation(com.nageoffer.onecoupon.framework.idempotent.NoMQDuplicateConsume)")public Object noMQRepeatConsume(ProceedingJoinPoint joinPoint) throws Throwable {NoMQDuplicateConsume noMQDuplicateConsume = getNoMQDuplicateConsumeAnnotation(joinPoint);String uniqueKey = noMQDuplicateConsume.keyPrefix() + SpELUtil.parseKey(noMQDuplicateConsume.key(), ((MethodSignature) joinPoint.getSignature()).getMethod(), joinPoint.getArgs());String absentAndGet = stringRedisTemplate.execute(RedisScript.of(LUA_SCRIPT, String.class),List.of(uniqueKey),IdempotentMQConsumeStatusEnum.CONSUMING.getCode(),String.valueOf(TimeUnit.SECONDS.toMillis(noMQDuplicateConsume.keyTimeout())));// 如果不为空证明已经有if (Objects.nonNull(absentAndGet)) {boolean errorFlag = IdempotentMQConsumeStatusEnum.isError(absentAndGet);log.warn("[{}] MQ repeated consumption, {}.", uniqueKey, errorFlag ? "Wait for the client to delay consumption" : "Status is completed");if (errorFlag) {throw new ServiceException(String.format("消息消费者幂等异常,幂等标识:%s", uniqueKey));}return null;}Object result;try {// 执行标记了消息队列防重复消费注解的方法原逻辑result = joinPoint.proceed();// 设置防重令牌 Key 过期时间,单位秒stringRedisTemplate.opsForValue().set(uniqueKey, IdempotentMQConsumeStatusEnum.CONSUMED.getCode(), noMQDuplicateConsume.keyTimeout(), TimeUnit.SECONDS);} catch (Throwable ex) {// 删除幂等 Key,让消息队列消费者重试逻辑进行重新消费stringRedisTemplate.delete(uniqueKey);throw ex;}return result;}/*** @return 返回自定义防重复消费注解*/public static NoMQDuplicateConsume getNoMQDuplicateConsumeAnnotation(ProceedingJoinPoint joinPoint) throws NoSuchMethodException {MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();Method targetMethod = joinPoint.getTarget().getClass().getDeclaredMethod(methodSignature.getName(), methodSignature.getMethod().getParameterTypes());return targetMethod.getAnnotation(NoMQDuplicateConsume.class);}

lua脚本解释

local key = KEYS[1] # 第一个 Key,即幂等唯一标识 uniqueKey
local value = ARGV[1] # 第一个参数,即初始化幂等消费状态,为消费中
local expire_time_ms = ARGV[2] # 第二个参数,即幂等 Key 过期时间return redis.call('SET', key, value, 'NX', 'GET', 'PX', expire_time_ms)

该脚本的主要作用是:在 Redis 中尝试以 NX 方式设置一个键,即如果键不存在,则设置新值,并返回设置之前的旧值,同时为该键设置过期时间(以毫秒为单位)。

获取到 Redis 里面的 Key 值后,可能会有三个流程执行:

absentAndGet 为空:代表消息是第一次到达,执行完 LUA 脚本后,会在 Redis 设置 Key 的 Value 值为 0,消费中状态。

absentAndGet 为 0:代表已经有相同消息到达并且还没有处理完,会通过抛异常的形式让 RocketMQ 重试。

absentAndGet 为 1:代表已经有相同消息消费完成,返回空表示不执行任何处理。

4.4.注册为 Spring Bean

另外可以看看另一篇基于分布式锁注解防重复提交

https://blog.csdn.net/sjsjsbbsbsn/article/details/145131305?spm=1001.2014.3001.5501

public class IdempotentConfiguration {/*** 防止消息队列消费者重复消费消息切面控制器*/@Beanpublic NoMQDuplicateConsumeAspect noMQDuplicateConsumeAspect(StringRedisTemplate stringRedisTemplate) {return new NoMQDuplicateConsumeAspect(stringRedisTemplate);}
}

4.5EL工具类

public class SpELUtil {/*** 校验并返回实际使用的 spEL 表达式** @param spEl spEL 表达式* @return 实际使用的 spEL 表达式*/public static Object parseKey(String spEl, Method method, Object[] contextObj) {List<String> spELFlag = ListUtil.of("#", "T(");Optional<String> optional = spELFlag.stream().filter(spEl::contains).findFirst();if (optional.isPresent()) {return parse(spEl, method, contextObj);}return spEl;}/*** 转换参数为字符串** @param spEl       spEl 表达式* @param contextObj 上下文对象* @return 解析的字符串值*/public static Object parse(String spEl, Method method, Object[] contextObj) {DefaultParameterNameDiscoverer discoverer = new DefaultParameterNameDiscoverer();ExpressionParser parser = new SpelExpressionParser();Expression exp = parser.parseExpression(spEl);String[] params = discoverer.getParameterNames(method);StandardEvaluationContext context = new StandardEvaluationContext();if (ArrayUtil.isNotEmpty(params)) {for (int len = 0; len < params.length; len++) {context.setVariable(params[len], contextObj[len]);}}return exp.getValue(context);}
}

5.实战使用

使用天机学堂项目来进行实战

5.1写入common模块

在这里插入图片描述

5.2使用

在这里插入图片描述

直接加上注解就可以

但是实际上这里不存在幂等问题,因为userId和courseId设置了唯一索引,所以这里不存在幂等性,不需要加上幂等注解

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

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

相关文章

Biotin sulfo-N-hydroxysuccinimide ester ;生物素磺基-N-羟基琥珀酰亚胺酯;生物素衍生物;190598-55-1

一、生物素及其衍生物的概述 生物素衍生物是指在生物素&#xff08;Vitamin H或B7&#xff09;分子基础上进行化学修饰得到的衍生化合物。这些衍生化合物在生物医学研究、临床诊断和药物开发等领域有着广泛的应用。 生物素&#xff08;Biotin&#xff09;是一种水溶性维生素&a…

Ubuntu如何安装redis服务?

环境&#xff1a; Ubuntu22.04 WSL2 问题描述&#xff1a; 如何安装redis服务&#xff1f; 解决方案&#xff1a; 1.在 Linux 上&#xff08;如 Ubuntu/Debian&#xff09;安装 1.通过包管理工具安装 Redis 服务器&#xff1a; sudo apt update sudo apt install redis…

Datawhale组队学习笔记task2——leetcode面试题

文章目录 写在前面Day5题目1.0112.路径总和解答2.0113路径总和II解答3.0101.对称二叉树解答 Day6题目1.0124.二叉树中的最大路径和解答2.0199.二叉树的右视图解答3.0226.翻转二叉树解答 Day7题目1.0105.从前序与中序遍历序列构造二叉树解答2.0098.验证二叉搜索树解答3.0110.平衡…

Slate文档编辑器-Node节点与Path路径映射

Slate文档编辑器-Node节点与Path路径映射 在之前我们聊到了slate中的Decorator装饰器实现&#xff0c;装饰器可以为我们方便地在编辑器渲染调度时处理range的渲染&#xff0c;这在实现搜索替换、代码高亮等场景非常有用。那么在这篇文章中&#xff0c;我们聊一下Node节点与Pat…

麒麟系统中删除权限不够的文件方法

在麒麟系统中删除权限不够的文件&#xff0c;可以尝试以下几种方法&#xff1a; 通过修改文件权限删除 打开终端&#xff1a;点击左下角的“终端”图标&#xff0c;或者通过搜索功能找到并打开终端 。定位文件&#xff1a;使用cd命令切换到文件所在的目录 。修改文件权限&…

Kotlin语言的正则表达式

Kotlin语言中的正则表达式 引言 正则表达式作为一种强大的文本处理工具&#xff0c;广泛应用于字符串匹配、数据验证、文本搜索等场景。在Kotlin语言中&#xff0c;正则表达式的应用同样得到了广泛关注。Kotlin不仅具备与Java相同的正则表达式功能优势&#xff0c;还提供了更…

Flask简介与安装以及实现一个糕点店的简单流程

目录 1. Flask简介 1.1 Flask的核心特点 1.2 Flask的基本结构 1.3 Flask的常见用法 1.3.1 创建Flask应用 1.3.2 路由和视图函数 1.3.3 动态URL参数 1.3.4 使用模板 1.4 Flask的优点 1.5 总结 2. Flask 环境创建 2.1 创建虚拟环境 2.2 激活虚拟环境 1.3 安装Flask…

RFID系统安全认证协议及防碰撞算法研究(RFID Security)

目录 1.摘要 2.引言 3.前人研究成果 3.1 RFID系统协议模型 3.2 RFID系统安全认证协议分类 3.3 RFID安全认证协议及其研究 3.3.1 超轻量级安全认证协议及其研究 3.3.2 轻量级安全认证协议及其研究 3.3.2 中量级安全认证协议及其研究 3.3.3 重量级安全认证协议及其研究…

Docker 实现MySQL 主从复制

一、拉取镜像 docker pull mysql:5.7相关命令&#xff1a; 查看镜像&#xff1a;docker images 二、启动镜像 启动mysql01、02容器&#xff1a; docker run -d -p 3310:3306 -v /root/mysql/node-1/config:/etc/mysql/ -v /root/mysql/node-1/data:/var/lib/mysql -e MYS…

Spring MVC:设置响应

目录 引言 1. 返回静态页面 1.1 Spring 默认扫描路径 1.2 RestController 1.2.1 Controller > 返回页面 1.2.2 ResponseBody 2. 返回 HTML 2.1 RequestMapping 2.1.1 produces(修改响应的 Content-Type) 2.1.2 其他属性 3. 返回 JSON 4. 设置状态码 4.1 HttpSer…

在Windows/Linux/MacOS C++程序中打印崩溃调用栈和局部变量信息

打印崩溃调用栈和局部变量信息的方法有所不同。以下是针对 Windows、Linux 和 MacOS 的示例代码。 Windows 在 Windows 上&#xff0c;可以使用 Windows API 来捕获异常并打印调用栈。 #include <windows.h> #include <DbgHelp.h> #include <stdio.h> #in…

基于python+Django+mysql鲜花水果销售商城网站系统设计与实现

博主介绍&#xff1a;黄菊华老师《Vue.js入门与商城开发实战》《微信小程序商城开发》图书作者&#xff0c;CSDN博客专家&#xff0c;在线教育专家&#xff0c;CSDN钻石讲师&#xff1b;专注大学生毕业设计教育、辅导。 所有项目都配有从入门到精通的基础知识视频课程&#xff…

提示词的艺术----AI Prompt撰写指南(个人用)

提示词的艺术 写在前面 制定提示词就像是和朋友聊天一样&#xff0c;要求我们能够清楚地表达问题。通过这个过程&#xff0c;一方面要不断练习提高自己地表达能力&#xff0c;另一方面还要锻炼自己使用更准确精炼的语言提出问题的能力。 什么样的提示词有用&#xff1f; 有…

MySQL管理事务处理

目录 1、事务处理是什么 2、控制事务处理 &#xff08;1&#xff09;事务的开始和结束 &#xff08;2&#xff09;回滚事务 &#xff08;3&#xff09;使用COMMIT &#xff08;4&#xff09;使用保留点 &#xff08;5&#xff09;结合存储过程的完整事务例子 3、小结 …

oneplus3t-lineageos-16.1编译-android9,

oneplus3t-lineageos-16.1编译-android9 oneplus3t 前提 救砖线刷 OnePlus3t android9 OTA卡刷 OnePlus3t android9 APatch root debian11(标准GNU工具集) arm 工具盘(chroot 风格rootfs, 含有 比如sshd 、gdb) : tinan/eadb.git 本仓库开发已经完毕,使用请直接从4.2开始…

Linux网络 序列化与反序列化

概念 序列化&#xff08;Serialization&#xff09;是将对象的状态信息转换为可以存储或传输的形式的过程。以下是关于序列化与反序列化的介绍&#xff1a; 序列化&#xff1a;将对象的状态信息转换为可以存储或传输的格式&#xff0c;通常是字节序列或文本格式。反序列化&am…

使用 spring boot 2.5.6 版本时缺少 jvm 配置项

2.5.6我正在使用带有版本和springfox-boot-starter版本的Spring Boot 项目3.0.0。我的项目还包括一个WebSecurityConfig扩展WebSecurityConfigurerAdapter并实现WebMvcConfigurer的类。但是&#xff0c;我面临的问题是指标在端点jvm_memory_usage_after_gc_percent中不可见/act…

python在财务领域的应用

财务岗位在处理数据时&#xff0c;经常会遇到一些复杂的场景&#xff0c;Excel 虽然功能强大&#xff0c;但在某些情况下可能无法高效或灵活地解决问题。以下是一些常见的、需要用编程&#xff08;如 Python、R 或 SQL&#xff09;来解决的数据问题&#xff1a; 1. 大规模数据处…

ZooKeeper 中的 ZAB 一致性协议与 Zookeeper 设计目的、使用场景、相关概念(数据模型、myid、事务 ID、版本、监听器、ACL、角色)

参考Zookeeper 介绍——设计目的、使用场景、相关概念&#xff08;数据模型、myid、事务 ID、版本、监听器、ACL、角色&#xff09; ZooKeeper 设计目的、特性、使用场景 ZooKeeper 的四个设计目标ZooKeeper 可以保证如下分布式一致性特性ZooKeeper 是一个典型的分布式数据一致…

Objective-C语言的数据类型

Objective-C数据类型详解 Objective-C是一种面向对象的编程语言&#xff0c;主要用于macOS和iOS应用程序的开发。作为C语言的超集&#xff0c;Objective-C继承了C语言的基本数据类型&#xff0c;同时也引入了一些独特的特性。本文将对Objective-C的各种数据类型进行详细的介绍…