spring schedule轻量级分布式调度方案--实现

• • 类结构图例
  • 代码
  • • • OnceOnTheSameTime
      • AbstractOnceService
      • MessageIdempotenceOnceService
      • OnceOnTheSameTimeAspect

接上文https://blog.csdn/weixin_43493520/article/details/107877905，介绍不再多说。

类结构图例

代码

OnceOnTheSameTime

/**
 * 业务语义的背景下，同一时刻，只能执行一次
 * 1. spring schedule，暂不支持分布式调度，提供轻量级分布式锁控制组件，解决这个问题
 * 2. 用户界面重复点击，创建同一条记录时，防重复，统一处理。
 * 3. 消息队列的幂等处理
 *
 **/
@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface OnceOnTheSameTime {

    /**
     * 定义once的有效期。
     */
    int timeoutInSeconds() default 30;

    /**
     * 定义重复的key: 使用SpEL语法。
     * 单参数举例：  name.concat('.').concat(age)
     * 多参数举例：  #args[0].name + '.' +#args[0].age
     */
    String keyDefinition() default "";
}

AbstractOnceService

 @Slf4j
public class AbstractOnceService {
    private static final String LOCK = "1";
    public static final String DOT = ".";
    public static final String ARGS = "args";

    @Value("${app.id}")
    private String appName;
    @Autowired
    private StringRedisTemplate redisTemplate;

    public Object aroundInvoke(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        Method method = methodSignature.getMethod();
        OnceOnTheSameTime anno = AnnotationUtils.getAnnotation(method, OnceOnTheSameTime.class);
        int timeoutInSeconds = anno.timeoutInSeconds();
        String key = constructKey(joinPoint, anno);

        boolean locked = lock(key, timeoutInSeconds);
        log.info("once aspect around invoked, redis key:{}, locked result:{}", key, locked);
        if (locked) {
            try {
                return joinPoint.proceed();
            } finally {
                unlock(key);
            }
        }

        throw new CommonException("repeated-execution", "重复执行");
    }

    private boolean lock(String key, int timeoutInSeconds) {
        return redisTemplate.opsForValue().setIfAbsent(key, LOCK, timeoutInSeconds, TimeUnit.SECONDS);
    }

    protected void unlock(String key) {
        redisTemplate.delete(key);
    }

    protected String constructKey(ProceedingJoinPoint joinPoint, OnceOnTheSameTime anno) {
        Object[] args = joinPoint.getArgs();

        if (args == null || args.length == 0) {
            return constructNoArgKey(joinPoint, anno);
        }

        if (args.length == 1) {
            return constructOneArgKey(joinPoint, anno, args[0]);
        }

        return constructMoreArgKey(joinPoint, anno, args);
    }

	/**
     * 构造多参数key
     *
     * @param joinPoint
     * @param anno
     * @param args
     * @return
     */
    private String constructMoreArgKey(ProceedingJoinPoint joinPoint, OnceOnTheSameTime anno, Object[] args) {

        StringBuilder keyBuilder = new StringBuilder(appName).append(DOT);
        EvaluationContext context = SimpleEvaluationContext.forReadOnlyDataBinding().build();
        context.setVariable(ARGS, args);

        SpelExpressionParser parser = new SpelExpressionParser();
        Expression expression = parser.parseExpression(anno.keyDefinition());
        String value = expression.getValue(context, String.class);
        keyBuilder.append(value);

        return keyBuilder.toString();
    }

    /**
     * 构造一个参数的key
     *
     * @param joinPoint
     * @param anno
     * @param arg
     * @return
     */
    private String constructOneArgKey(ProceedingJoinPoint joinPoint, OnceOnTheSameTime anno, Object arg) {
        StringBuilder keyBuilder = new StringBuilder(appName).append(DOT);

        SpelExpressionParser parser = new SpelExpressionParser();
        Expression expression = parser.parseExpression(anno.keyDefinition());
        Object value = expression.getValue(arg);
        keyBuilder.append(value);

        return keyBuilder.toString();
    }

    /**
     * 构造无参数的key
     *
     * @param joinPoint
     * @param anno
     * @return
     */
    private String constructNoArgKey(ProceedingJoinPoint joinPoint, OnceOnTheSameTime anno) {
        StringBuilder keyBuilder = new StringBuilder(appName).append(DOT);

        Object target = joinPoint.getTarget();
        //得到其方法签名
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        Method method = methodSignature.getMethod();

        String keyDefination = anno.keyDefinition();
        //处理无参数key定义
        String keySuffix = StringUtils.isBlank(keyDefination) ? target.getClass().getSimpleName() + "." + method.getName() : keyDefination;
        return keyBuilder.append(keySuffix).toString();
    }


}

MessageIdempotenceOnceService

/**
 * 消息队列的幂等处理*
 **/
@Component
@Slf4j
public class MessageIdempotenceOnceService extends AbstractOnceService {

    @Override
    protected void unlock(String key) {
        // do nothing..
    }

}

}

OnceOnTheSameTimeAspect

@Component
@Aspect
@Slf4j
public class OnceOnTheSameTimeAspect {

    @Autowired
    private CreateOperateOnceService createOperateOnceService;

    @Autowired
    private DistributeScheduleOnceService distributeScheduleOnceService;

    @Autowired
    private MessageIdempotenceOnceService messageIdempotenceOnceService;

    /**
     * execution(* xx.web.controller..*.create*(..))
     * execution(* set*(..))
     */
    @Pointcut("execution(* xx.web.controller..*.create*(..)) && @annotation(xx.annotation.OnceOnTheSameTime)")
    public void createCondition() {
    }

    /**
     * target方法如果只有一个参数，直接使用SpEL root对象取值即可；
     * target方法如果有多个参数，表达式中使用#args[0,1]来取值
     *
     * @param joinPoint
     * @throws Throwable
     */
    @Around("createCondition()")
    public Object createAround(ProceedingJoinPoint joinPoint) throws Throwable {
        return createOperateOnceService.aroundInvoke(joinPoint);
    }


    @Pointcut("bean(scheduleConfig) && @annotation(xx.annotation.OnceOnTheSameTime)")
    public void scheduleCondition() {
    }

    @Around("scheduleCondition()")
    public Object scheduleAround(ProceedingJoinPoint joinPoint) throws Throwable {
        return distributeScheduleOnceService.aroundInvoke(joinPoint);
    }


    @Pointcut("execution(* xx.mq..*Consumer.*(..)) && @annotation(xx.annotation.OnceOnTheSameTime)")
    public void mqCondition() {
    }

    @Around("mqCondition()")
    protected Object mqAround(ProceedingJoinPoint joinPoint) throws Throwable {
        return messageIdempotenceOnceService.aroundInvoke(joinPoint);
    }
}