使用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分布式锁的使用场景还有很多，经常用在高并发环境下来保证程序的正确性。