Redis分布式锁(一):基于AOP和Redis实现的简易版分布式锁

平时在项目中经常会遇到并发问题,理论上有些方法只应该被调用一次,但由于并发而被重复调用,导致出现系统问题。而在我们的项目中,最容易发生这种情况的是,推送消息的服务。无论是短信还是app推送,或者是公众号的推送,因为并发问题而导致的重复推送,肯定是不被允许的。

所以我准备通过分布式锁去解决这个问题。

分布式锁一般解决如下两类问题:
1、效率性问题,比如重复发短信,重复生成同样的订单等。
2、正确性问题,比如在某一个请求扣款的同时不允许其他请求同时去扣款。

当然,题目中写了简易版是因为,这次的分布式锁只准备解决效率性的问题,而不解决正确性的问题。

之所以基于Redis,是因为Redis作为一款基本上互联网公司都会必备的缓存数据库,简单好用,入门门槛较低。

而基于AOP实现,是因为用注解方式侵入性更低,基本上不需要修改原生代码,只需要加注解就行。

如下是简易版的分布式锁的实现方式:

首先我们需要2个注解,用于方法的LockAnnotation和用于参数的LockParamAnnotation

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD, ElementType.TYPE})
public @interface LockAnnotation {

    /**
     * 加锁的key的域,用于前缀识别
     *
     * @return
     */
    String lockField() default "";

    /**
     * 锁自动释放时间
     *
     * @return
     */
    int lockTime() default 3;
}

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.PARAMETER, ElementType.TYPE})
@Documented
public @interface LockParamAnnotation {
    /**
     * 参数的域,用于表明参数的业务场景,例如orderSn等
     *
     * @return
     */
    String value() default "";
}

LockAnnotation用于加在需要分布式锁的方法上,表明这是一个需要加分布式锁的方法,来引入切面,lockField用于redis的key的前缀识别,而lockTime是防止死锁而添加的锁的默认失效时间。
而LockParamAnnotation需要和LockAnnotation配套使用,需要加在方法的参数上,用于该方法的redis的唯一的key的组成。

最后是切面的实现LockAspect

@Component
@Aspect
public class LockAspect {

    private static Logger logger = LoggerFactory.getLogger(LockAspect.class);

    @Resource
    private CacheUtils cacheUtils;

    @Pointcut("@annotation(com.xiaoyao.service.lock.LockAnnotation)")
    public void lockAspect() {

    }

    @Around(value = "lockAspect()")
    public Object lockAround(ProceedingJoinPoint joinPoint) throws Throwable {
        //增强处理的目标对象的类
        Class targetClass = joinPoint.getTarget().getClass();
        //目标对象的所有方法
        Method[] methods = targetClass.getMethods();
        Object[] args = joinPoint.getArgs();

        Signature signature = joinPoint.getSignature();
        Optional<Method> optional = Lists.newArrayList(methods).stream().filter(m -> Objects.equals(m.getName(), signature.getName())).findFirst();
        Object returnObject = null;
        List<String> annotationParamList = Lists.newArrayList();
        if (optional.isPresent()) {
            Method method = optional.get();
            LockAnnotation lockAnnotation = method.getAnnotation(LockAnnotation.class);
            Annotation[][] parameterAnnotations = method.getParameterAnnotations();
            final int[] index = {0};
            Lists.newArrayList(parameterAnnotations).forEach(param -> {
                Optional annotationOptional = Lists.newArrayList(param).stream().filter(m -> Objects.equals(m.annotationType(), LockParamAnnotation.class)).findFirst();
                if (annotationOptional.isPresent()) {
                    LockParamAnnotation annotation = (LockParamAnnotation) annotationOptional.get();
                    annotationParamList.add(annotation.value() + ":" + args[index[0]].toString());
                    return;
                }
                index[0]++;
            });
            String lockKey = String.join("#", annotationParamList);
            String lockField = lockAnnotation.lockField();
            int lockTime = lockAnnotation.lockTime();
            try {
                String randomValue = UUID.randomUUID().toString();
                if (!"OK".equals(cacheUtils.setLock(lockField, lockKey, randomValue, lockTime))) {
                    if (logger.isInfoEnabled()) {
                        logger.info("无法获得锁,方法名为{},参数为{}", joinPoint.getSignature(), String.join("-", Lists.newArrayList(args).stream().map(obj -> JSONObject.toJSONString(ObjectUtils.defaultIfNull(obj, "null"))).collect(Collectors.toList())));
                    }
                    return null;
                }
                if (logger.isInfoEnabled()) {
                    logger.info("可以获得锁,方法名为{},参数为{}", joinPoint.getSignature(), String.join("-", Lists.newArrayList(args).stream().map(obj -> JSONObject.toJSONString(ObjectUtils.defaultIfNull(obj, "null"))).collect(Collectors.toList())));
                }
                returnObject = joinPoint.proceed(args);
            } finally {
                cacheUtils.delLock(lockField, lockKey, randomValue);
            }
        }
        return returnObject;
    }
}

其中setLock是通过Redis的set指令实现的,其中[NX|XX]选择NX,然后设置过期时间。
而delLock是通过Redis的eval指令去执行lua脚本实现的,其中script代码如下:

String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

以上获取锁和释放锁的过程经过一次改良,原本不太完美的的代码如下:

try {
    if (cacheUtils.incr(lockField, lockKey) > 1) {
        if (logger.isDebugEnabled()) {
            logger.debug("无法获得锁,方法名为{},参数为{}", joinPoint.getSignature(), String.join("-", Lists.newArrayList(args).stream().map(obj -> JSONObject.toJSONString(ObjectUtils.defaultIfNull(obj, "null"))).collect(Collectors.toList())));
       }
        return null;
    }
    cacheUtils.expireKey(lockField, lockKey, lockTime);
    if (logger.isDebugEnabled()) {
        logger.debug("可以获得锁,方法名为{},参数为{}", joinPoint.getSignature(), String.join("-", Lists.newArrayList(args).stream().map(obj -> JSONObject.toJSONString(ObjectUtils.defaultIfNull(obj, "null"))).collect(Collectors.toList())));
    }
    returnObject = joinPoint.proceed(args);
} finally {
    if (cacheUtils.ttl(lockField, lockKey) > 0) {
        cacheUtils.del(lockField, lockKey);
    }
}

主要区别在于有两点。
1.之前的创建锁和给锁设置有效时间是分二步进行的,存在风险。万一在创建锁之后宕机了,那么将彻底死锁。改良后,将创建锁和给锁设置有效时间合并为一步SetNX,就避免了这个问题。

2.之前的释放锁的过程,只判断了ttl,但没法保证释放锁的时候,自己依然是锁的持有者。如果不加以判断直接释放锁,就会出现误删除其它请求创建的锁的情况。改良后,通过生成一个随机数,然后在释放锁的时候通过lua脚本去执行,先获取锁的值,然后判断释放的锁是否是当初该请求创建的锁,如果是,则释放锁。

if (randomValue.equals(cacheUtils.get(lockField, lockKey))) {
      cacheUtils.del(lockField, lockKey);
}

之所以没有采用上述的先get再del的操作,而是采用lua脚本是因为需要保持操作的原子性。
假设采用上述的方式,则get和del是分步执行的。那么如果要求a在执行del操作之前,万一因为其他原因导致没有及时del,此时锁过期自动释放了,这时请求b发现可以创建锁,就创建了锁。然后请求a突然又恢复正常去释放锁,但此时锁的持有者是请求b,请求a误删了请求b持有的锁。
这个问题比较复杂,因为Redis没有get和del合二为一的操作,要解决该问题只能通过lua脚本将这两个操作合二为一,一起执行才行。

原理介绍完了,接下来介绍一下如何使用。

@LockAnnotation(lockField = "lockKey",lockTime =1)
public void lock(String action,@LockParamAnnotation("orderSn") String orderSn){
    
}

简单易用,大家可以自己用多线程试用一下

后续已做一次优化,详见下文基于AOP和Redis实现的简易版分布式锁(二)

如有更好的建议,欢迎一起来沟通。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容