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();
    //如果方法所处的类是一个interface
    if (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() : #instanceId
        Expression 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的创建，每一个知识点都够我喝一壶了，所以大家如果想要了解这些知识的细节，请自行搜索去查想要了解的内容，如果你懒得查，也可以问我，当然我也懒，回不回答就看我心情了，哼，我外号就叫不高兴，所以大家看着办吧。