AOP在PowerJob中的使用,缓存锁保证并发安全,知识细节全总结

2023-10-07 08:59

本文主要是介绍AOP在PowerJob中的使用,缓存锁保证并发安全,知识细节全总结,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

这是一篇简简单单的文章,需要你简简单单看一眼就好,如果有不明白的地方,欢迎留言讨论。

 

在之前的文章中出现过一次AOP的使用,就是在运行任务之前,需要判断一下,触发该任务执行的server,是不是数据库中对应任务所在app的直接server,使用的是注解@DesignateServer,本篇文章是从另一个注解,再一次顺一遍AOP的使用,而且本篇文章的注解,再一次用到了可重入锁ReentrantLock,这个也是之前的文章中说的内容,可以再熟悉一遍,本篇文章的入口就是注解——UseCacheLock。

从名字来看,该注解是一个使用缓存时的一个锁,该类位于tech.powerjob.server.core.lock包下,用来修饰方法,在运行时执行的,源码如下:

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface UseCacheLock {String type();String key();int concurrencyLevel();
}

type:从使用的代码处得出,目前只有两种一种是processJobInstance,另一种是processWfInstance

key:主要是任务id或者任务实例id,还有工作流id。其中任务id或者任务实例id的选取,是通过一个表达式来判断得出的。

concurrencyLevel:缓存要用到的字段,允许同时并发执行的写操作数。

UseCacheLock 的使用场景

该注解在powerjob中一共使用了8次,其中2次出现在任务的派发,6次出现在工作流的操作中,这一次就选在任务的派发,来讲一下该注解的使用场景。

@UseCacheLock(type = "processJobInstance", key = "#jobInfo.getMaxInstanceNum() > 0 || T(tech.powerjob.common.enums.TimeExpressionType).FREQUENT_TYPES.contains(#jobInfo.getTimeExpressionType()) ? #jobInfo.getId() : #instanceId", concurrencyLevel = 1024)
public void dispatch(JobInfoDO jobInfo, Long instanceId) {... ...
}

方法内部的代码不重要,主要是来看方法上面的注解,里面的三个关键字分别是

processJobInstance

#jobInfo.getMaxInstanceNum() > 0 || T(tech.powerjob.common.enums.TimeExpressionType).FREQUENT_TYPES.contains(#jobInfo.getTimeExpressionType()) ? #jobInfo.getId() : #instanceId

1024

第一个和最后一个没什么好说的,主要说一说中间那一条长长的表达式,该表达式通过解读,就是判断最大同时运行任务数是否大于0,以及任务的时间表达式类型是不是FIX_RATE或者FIX_DELAY。这一表达式可以说,除非人为的将MaxInstanceNum设置为0,否则该条数据默认值就是1,也就是说这个表达式,不负责任的说,99.99%都是真,也就是说都会使用JobId作为key值。

按照代码来看,就是当任务在派发的时候,会使用到该注解,为的是防止该方法同时运行派发同一个任务,如果是同时派发两个不同的任务,就不会有影响,毕竟在派发的过程中涉及到了对任务实例的数据修改,如果两个同时进行,确实会产生问题。

UseCacheLock 的AOP处理

处理该注解的类个该类在同一个包,处理的源代码如下所示:

@Around(value = "@annotation(useCacheLock)")public Object execute(ProceedingJoinPoint point, UseCacheLock useCacheLock) throws Throwable {Cache<String, ReentrantLock> lockCache = lockContainer.computeIfAbsent(useCacheLock.type(), ignore -> {int concurrencyLevel = useCacheLock.concurrencyLevel();log.info("[UseSegmentLockAspect] create Lock Cache for [{}] with concurrencyLevel: {}", useCacheLock.type(), concurrencyLevel);return CacheBuilder.newBuilder().initialCapacity(300000).maximumSize(500000).concurrencyLevel(concurrencyLevel).expireAfterWrite(30, TimeUnit.MINUTES).build();});final Method method = AOPUtils.parseMethod(point);Long key = AOPUtils.parseSpEl(method, point.getArgs(), useCacheLock.key(), Long.class, 1L);final ReentrantLock reentrantLock = lockCache.get(String.valueOf(key), ReentrantLock::new);long start = System.currentTimeMillis();reentrantLock.lockInterruptibly();try {long timeCost = System.currentTimeMillis() - start;if (timeCost > SLOW_THRESHOLD) {final SlowLockEvent slowLockEvent = new SlowLockEvent().setType(SlowLockEvent.Type.LOCAL).setLockType(useCacheLock.type()).setLockKey(String.valueOf(key)).setCallerService(method.getDeclaringClass().getSimpleName()).setCallerMethod(method.getName()).setCost(timeCost);monitorService.monitor(slowLockEvent);log.warn("[UseSegmentLockAspect] wait lock for method({}#{}) cost {} ms! key = '{}', args = {}, ", method.getDeclaringClass().getSimpleName(), method.getName(), timeCost,key,JSON.toJSONString(point.getArgs()));}return point.proceed();} finally {reentrantLock.unlock();}
}

代码看着挺长的,但是内容其实没有多少,可以一步一步拆开来看。

缓存的创建

第一步通过type来获取缓存,从文章开头我们知道,这个type就两个类型,processJobInstance就是用来派发任务的,processWfInstance就是用来操作工作流任务的,该代码里面就是processJobInstance,如果缓存存在,直接拿来用,如果不存在,则创建缓存,来看一眼创建缓存的代码:

CacheBuilder.newBuilder().initialCapacity(300000).maximumSize(500000).concurrencyLevel(concurrencyLevel).expireAfterWrite(30, TimeUnit.MINUTES).build();

这个代码的大意就是创建一个有如下属性的缓存,缓存有效时间是30分钟(expireAfterWrite(30, TimeUnit.MINUTES)),这就像鱼有7秒记忆一样,这个缓存只能记录30分钟,过期失效。缓存的最大条目数是50万(maximumSize(500000))。指定用于缓存的hash table最低总规模是300000,允许同时并发操作数是concurrencyLevel,也就是传进来的1024.

key值的获取

第二步就是获取key值,该值主要是为了获取可重入锁用的,获取该值的源代码如下所示:

final Method method = AOPUtils.parseMethod(point);
Long key = AOPUtils.parseSpEl(method, point.getArgs(), useCacheLock.key(), Long.class, 1L);

从这个代码可以看到,用到了AOPUtil这个工具类的两个方法,第一个方法是解析出当前的方法,第二个是获取key值,这个AOPUtil在tech.powerjob.server.common.utils包下。解析方法的源码如下,备注解释各代码的目的:

public static Method parseMethod(ProceedingJoinPoint joinPoint) {//获取接入点的签名,此处必须是方法的签名,否则会报异常Signature pointSignature = joinPoint.getSignature();if (!(pointSignature instanceof  MethodSignature)) {throw new IllegalArgumentException("this annotation should be used on a method!");}//强转成方法的签名MethodSignature signature = (MethodSignature) pointSignature;//获取方法Method method = signature.getMethod();//如果方法所处的类是一个interfaceif (method.getDeclaringClass().isInterface()) {try {//通过IoC容器获取目标对象,然后再获取对象的方法method = joinPoint.getTarget().getClass().getDeclaredMethod(pointSignature.getName(), method.getParameterTypes());} catch (SecurityException | NoSuchMethodException e) {ExceptionUtils.rethrow(e);}}return method;
}

获取到了方法之后,就是获取key值,源代码如下,备注解释各代码的目的:

public static <T> T parseSpEl(Method method, Object[] arguments, String spEl, Class<T> clazz, T defaultResult) {//获取到方法的参数值类型String[] params = discoverer.getParameterNames(method);assert params != null;//创建数据上下文EvaluationContext context = new StandardEvaluationContext();for (int len = 0; len < params.length; len++) {//将param[len] = arguments[len]context.setVariable(params[len], arguments[len]);}try {//执行表达式,也就是前面#jobInfo.getMaxInstanceNum() > 0 || T(tech.powerjob.common.enums.TimeExpressionType).FREQUENT_TYPES.contains(#jobInfo.getTimeExpressionType()) ? #jobInfo.getId() : #instanceIdExpression expression = parser.parseExpression(spEl);//返回表达式执行的结果,以clazz设置的类型返回return expression.getValue(context, clazz);} catch (Exception e) {log.error("[AOPUtils] parse SpEL failed for method[{}], please concat @tjq to fix the bug!", method.getName(), e);return defaultResult;}
}

经过以上两步,key值就获取完毕了

加锁

加锁的源代码如下所示,就是如果缓存里面保存了锁,就直接拿到,如果没有,就new一个出来,然后就启动锁,

那两条时间主要是记录加锁的时间,如果时间过长就要记录一条日志,记录加锁慢时的任务信息。

final ReentrantLock reentrantLock = lockCache.get(String.valueOf(key), ReentrantLock::new);
long start = System.currentTimeMillis();
reentrantLock.lockInterruptibly();
try {long timeCost = System.currentTimeMillis() - start;... ...
}
... ...

加锁结束之后,就可以执行注解修饰的方法了,执行就是下面这一行:

point.proceed();

执行结束之后,将锁打开就OK了。

总结

 

本篇文章涉及的知识主要是AOP的使用,可重入锁的使用,IoC容器相关,Spring的表达式的使用,缓存Cache的创建,每一个知识点都够我喝一壶了,所以大家如果想要了解这些知识的细节,请自行搜索去查想要了解的内容,如果你懒得查,也可以问我,当然我也懒,回不回答就看我心情了,哼,我外号就叫不高兴,所以大家看着办吧。

这篇关于AOP在PowerJob中的使用,缓存锁保证并发安全,知识细节全总结的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/157341

相关文章

使用Python实现高效的端口扫描器

《使用Python实现高效的端口扫描器》在网络安全领域,端口扫描是一项基本而重要的技能,通过端口扫描,可以发现目标主机上开放的服务和端口,这对于安全评估、渗透测试等有着不可忽视的作用,本文将介绍如何使... 目录1. 端口扫描的基本原理2. 使用python实现端口扫描2.1 安装必要的库2.2 编写端口扫

使用Python实现操作mongodb详解

《使用Python实现操作mongodb详解》这篇文章主要为大家详细介绍了使用Python实现操作mongodb的相关知识,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录一、示例二、常用指令三、遇到的问题一、示例from pymongo import MongoClientf

SQL Server使用SELECT INTO实现表备份的代码示例

《SQLServer使用SELECTINTO实现表备份的代码示例》在数据库管理过程中,有时我们需要对表进行备份,以防数据丢失或修改错误,在SQLServer中,可以使用SELECTINT... 在数据库管理过程中,有时我们需要对表进行备份,以防数据丢失或修改错误。在 SQL Server 中,可以使用 SE

使用Python合并 Excel单元格指定行列或单元格范围

《使用Python合并Excel单元格指定行列或单元格范围》合并Excel单元格是Excel数据处理和表格设计中的一项常用操作,本文将介绍如何通过Python合并Excel中的指定行列或单... 目录python Excel库安装Python合并Excel 中的指定行Python合并Excel 中的指定列P

浅析Rust多线程中如何安全的使用变量

《浅析Rust多线程中如何安全的使用变量》这篇文章主要为大家详细介绍了Rust如何在线程的闭包中安全的使用变量,包括共享变量和修改变量,文中的示例代码讲解详细,有需要的小伙伴可以参考下... 目录1. 向线程传递变量2. 多线程共享变量引用3. 多线程中修改变量4. 总结在Rust语言中,一个既引人入胜又可

golang1.23版本之前 Timer Reset方法无法正确使用

《golang1.23版本之前TimerReset方法无法正确使用》在Go1.23之前,使用`time.Reset`函数时需要先调用`Stop`并明确从timer的channel中抽取出东西,以避... 目录golang1.23 之前 Reset ​到底有什么问题golang1.23 之前到底应该如何正确的

SpringBoot实现动态插拔的AOP的完整案例

《SpringBoot实现动态插拔的AOP的完整案例》在现代软件开发中,面向切面编程(AOP)是一种非常重要的技术,能够有效实现日志记录、安全控制、性能监控等横切关注点的分离,在传统的AOP实现中,切... 目录引言一、AOP 概述1.1 什么是 AOP1.2 AOP 的典型应用场景1.3 为什么需要动态插

详解Vue如何使用xlsx库导出Excel文件

《详解Vue如何使用xlsx库导出Excel文件》第三方库xlsx提供了强大的功能来处理Excel文件,它可以简化导出Excel文件这个过程,本文将为大家详细介绍一下它的具体使用,需要的小伙伴可以了解... 目录1. 安装依赖2. 创建vue组件3. 解释代码在Vue.js项目中导出Excel文件,使用第三

Kubernetes常用命令大全近期总结

《Kubernetes常用命令大全近期总结》Kubernetes是用于大规模部署和管理这些容器的开源软件-在希腊语中,这个词还有“舵手”或“飞行员”的意思,使用Kubernetes(有时被称为“... 目录前言Kubernetes 的工作原理为什么要使用 Kubernetes?Kubernetes常用命令总

Linux alias的三种使用场景方式

《Linuxalias的三种使用场景方式》文章介绍了Linux中`alias`命令的三种使用场景:临时别名、用户级别别名和系统级别别名,临时别名仅在当前终端有效,用户级别别名在当前用户下所有终端有效... 目录linux alias三种使用场景一次性适用于当前用户全局生效,所有用户都可调用删除总结Linux