在之前的文章中出现过一次AOP的使用#xff0c;就是在运行任务之前#xff0c;需要判断一下#xff0c;触发该任务执行的server#xff0c;是不是数…

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

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

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

Retention(RetentionPolicy.RUNTIME)

public

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

key主要是任务id或者任务实例id还有工作流id。

其中任务id或者任务实例id的选取是通过一个表达式来判断得出的。

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

UseCacheLock

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

UseCacheLock(type

T(tech.powerjob.common.enums.TimeExpressionType).FREQUENT_TYPES.contains(#jobInfo.getTimeExpressionType())

#jobInfo.getId()

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

processJobInstance

T(tech.powerjob.common.enums.TimeExpressionType).FREQUENT_TYPES.contains(#jobInfo.getTimeExpressionType())

#jobInfo.getId()

第一个和最后一个没什么好说的主要说一说中间那一条长长的表达式该表达式通过解读就是判断最大同时运行任务数是否大于0以及任务的时间表达式类型是不是FIX_RATE或者FIX_DELAY。

这一表达式可以说除非人为的将MaxInstanceNum设置为0否则该条数据默认值就是1也就是说这个表达式不负责任的说99.99%都是真也就是说都会使用JobId作为key值。

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

UseCacheLock

annotation(useCacheLock))public

Object

lockContainer.computeIfAbsent(useCacheLock.type(),

ignore

useCacheLock.concurrencyLevel();log.info([UseSegmentLockAspect]

create

CacheBuilder.newBuilder().initialCapacity(300000).maximumSize(500000).concurrencyLevel(concurrencyLevel).expireAfterWrite(30,

TimeUnit.MINUTES).build();});final

Method

AOPUtils.parseMethod(point);Long

key

lockCache.get(String.valueOf(key),

start

System.currentTimeMillis();reentrantLock.lockInterruptibly();try

{long

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

method.getDeclaringClass().getSimpleName(),

method.getName(),

timeCost,key,JSON.toJSONString(point.getArgs()));}return

point.proceed();}

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

CacheBuilder.newBuilder().initialCapacity(300000).maximumSize(500000).concurrencyLevel(concurrencyLevel).expireAfterWrite(30,

这个代码的大意就是创建一个有如下属性的缓存缓存有效时间是30分钟expireAfterWrite(30,

TimeUnit.MINUTES)这就像鱼有7秒记忆一样这个缓存只能记录30分钟过期失效。

缓存的最大条目数是50万maximumSize(500000)。

指定用于缓存的hash

table最低总规模是300000允许同时并发操作数是concurrencyLevel也就是传进来的1024.

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

final

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

解析方法的源码如下备注解释各代码的目的

public

parseMethod(ProceedingJoinPoint

joinPoint)

{//获取接入点的签名此处必须是方法的签名否则会报异常Signature

pointSignature

method!);}//强转成方法的签名MethodSignature

signature

signature.getMethod();//如果方法所处的类是一个interfaceif

(method.getDeclaringClass().isInterface())

{try

{//通过IoC容器获取目标对象然后再获取对象的方法method

joinPoint.getTarget().getClass().getDeclaredMethod(pointSignature.getName(),

catch

{ExceptionUtils.rethrow(e);}}return

method;

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

public

discoverer.getParameterNames(method);assert

params

null;//创建数据上下文EvaluationContext

context

StandardEvaluationContext();for

(int

arguments[len]context.setVariable(params[len],

arguments[len]);}try

{//执行表达式也就是前面#jobInfo.getMaxInstanceNum()

T(tech.powerjob.common.enums.TimeExpressionType).FREQUENT_TYPES.contains(#jobInfo.getTimeExpressionType())

#jobInfo.getId()

parser.parseExpression(spEl);//返回表达式执行的结果以clazz设置的类型返回return

clazz);}

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

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

final

lockCache.get(String.valueOf(key),

ReentrantLock::new);

reentrantLock.lockInterruptibly();

try

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