redis分布式锁自定义注解实现分布式环境下的定时任务

使用redis锁控制定时任务的意义

 　　有一次在开发一个获取审批状态结果的接口时，用了一个定时任务定时去拉取的操作，在这个系统中，我没有直接接入我们的xxl-job，因为我想换一种实现方式来试一下，同时业务对定时任务的需求不高，所以我打算尝试使用@Scheduled来实现。

 　　将cron表达式的值配置在Apollo上，实现可以动态刷新的机制。这样定时任务确实是可以跑起来的，但是一般我们的应用都是一个分布式部署的状态，一个应用部署了多台服务器，那么这里就出现了一个问题：就是当每台机器上的定时任务都被触发了，同时去拉取状态以及更改数据，那么这里就可能导致将数据更改错误，引起不必要的问题，所以我们得通过手段来控制，使的到任务触发时间，只需要一台应用服务器去触发任务就好了。

此处的Apollo是携程开源的一款分布式的配置中心

单机下使用@Scheduled没有问题哈

代码实现

redis配置

 　　redis的基本配置这个自己在网上找一下就可以了。主要是用到了RedisTemlate，下面主要讲解redis锁定时任务的实现，这里是将redis获取锁的过程通过AOP做成了一个公用的注解组件，提高系统的扩展性以及共用性。

定义RedisUtils工具类

 　　redis的工具类网上很多，我这里也是摘抄的，主要看加锁的方法，其他的方法就不粘贴了。

package com.anzz.anzzdemo.redis;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Component
public final class RedisUtils {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;


	/**
     * 加强版分布式锁
     * @param key
     * @param lockExpire
     * @return
     */
    @SuppressWarnings({"unchecked", "rawtypes"})
    public boolean lock(String key, Long lockExpire) {
        // lambda 表达式
        return (Boolean) redisTemplate.execute((RedisCallback) connection -> {
            // 加1是因为时间戳的精度为1s，为了避免误差，所以增加1s
            long expireAt = System.currentTimeMillis() + lockExpire + 1;
            Boolean acquire = connection.setNX(key.getBytes(), String.valueOf(expireAt).getBytes());
            if (acquire) {
                return true;
            } else {
                byte[] value = connection.get(key.getBytes());
                if (Objects.nonNull(value) && value.length > 0) {
                    long expireTime = Long.parseLong(new String(value));
                    // 如果锁已经过期
                    if (expireTime < System.currentTimeMillis()) {
                        // 重新加锁，防止死锁
                        byte[] oldValue = connection.getSet(key.getBytes(), String.valueOf(System.currentTimeMillis() + lockExpire + 1).getBytes());
                        return Long.parseLong(new String(oldValue)) < System.currentTimeMillis();
                    }
                }
            }
            return false;
        });
    }
}

定义一个RedisLock自定义注解

package com.anzz.anzzdemo.redis;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;


@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RedisLock {


    /**
     * 重试机制：是否需要重新尝试获取锁
     * @return
     */
    boolean needRetry() default true;


    /**
     * 没有获取到锁是是否需要抛出异常
     * @return
     */
    boolean notLockNeedThrowException() default true;

    /**
     * 提示
     * @return
     */
    String errorMsg() default "系统繁忙，请稍后重试";


}

实现切面RedisLockAspect

 　　注解切面，在这里，获取到加了注解的方法，先获取锁，获取到锁才执行真正的方法。

package com.anzz.anzzdemo.redis;

import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;



@Slf4j
@Aspect
@Component
public class RedisLockAspect {

    @Autowired
    private RedisUtils redisUtils;

    /**
     * 重试次数：这里可以配置在配置中心上，此处仅为说明
     */
    private int retryCount = 5;

    /**
     * 重试时间
     */
    private long retryTime = 200;

    /**
     * 过期时间
     */
    private long redisLockExpire;

    @Around("@annotation(RedisLock)")
    public Object process(ProceedingJoinPoint point) throws Throwable {
        String redisKey = getRedisKey(point);
        log.info("### 开始进入RedisLock切面，key:{}", redisKey);
        RedisLock redisLock = getRedisLockInfo(point);
        boolean needRetry = redisLock.needRetry();
        if (needRetry) {
            int rCount = 1;
            while (!redisUtils.lock(redisKey, redisLockExpire)) {
                if (rCount > retryCount) {
                    if (redisLock.notLockNeedThrowException()) {
                        throw new RuntimeException(redisLock.errorMsg());
                    }
                    return null;
                }
                // 线程等待
                Thread.sleep(retryTime);
                rCount++;
            }
        } else {
            if (!redisUtils.lock(redisKey, redisLockExpire)) {
                if (redisLock.notLockNeedThrowException()) {
                    throw new RuntimeException(redisLock.errorMsg());
                }
                return null;
            }
        }
        // 获取锁成功
        try {
            Object[] args = point.getArgs();
            return point.proceed(args);
        } finally {
            // 释放锁
            redisUtils.del(redisKey);
        }
    }


    /**
     * 根据自身业务设置redis的key
     * @param point
     * @return
     */
    private String getRedisKey(ProceedingJoinPoint point) {
        String sufix_key = "09090";
        return "XX-RE" + point.getSignature().getDeclaringTypeName() + "-" + point.getSignature().getName() + sufix_key;
    }


    /**
     * 获取注解及方法信息
     * @param point
     * @return
     * @throws NoSuchMethodError
     */
    private RedisLock getRedisLockInfo(ProceedingJoinPoint point) throws NoSuchMethodException {
        String methodName = point.getSignature().getName();
        Class<?> classTarget = point.getTarget().getClass();
        Class<?>[] params = ((MethodSignature) point.getSignature()).getParameterTypes();
        Method method = classTarget.getMethod(methodName, params);
        return method.getAnnotation(RedisLock.class);
    }

}

总结

 　　至此，通过redis分布式实现定时任务就完成啦，redis分布式锁的使用场景还有很多，经常用在高并发环境下来保证程序的正确性。