Redis之分布式锁实现

最近在工作中使用到了分布式锁，特此总结一下便于加深自己理解。将会从如下的几个方面进行讲述。

1. 分布式锁的实现本质上就是基于redis的命令去实现的，setnx按照官方的定义是这样的：

Redis SETNX 命令用于将 Redis 的 KEY 的值设为 value ，当且仅当 KEY 不存在。若给定的 KEY 已经存在，则 SETNX 不做任何动作。

那按照这个定义去试一试会得到如下的结果:

2. 实现一个简单的分布式锁的代码如下

    public String test(String userName){
        
        // 加锁
        Boolean result = Optional.ofNullable(stringRedisTemplate.opsForValue().setIfAbsent(userName, String.valueOf(3))).orElse(Boolean.FALSE);
        
        if (!result){
            // 执行加锁失败的逻辑
            return null;
        }
        
        try {
            //执行业务逻辑
        }catch (Exception e){
            
        }finally {
            // 释放锁
            stringRedisTemplate.delete(userName);
        }
        
        return null;
    }

上述的分布式锁虽然实现了但是会存在如下的问题：
a.会出现死锁，出现死锁的原因是这样的。当还没有来得及执行finally中逻辑时。应用宕机了。导致锁没有办法被释放。那这个不是挺简单的吗，加个TTL不就可以了吗。我们看看在假如TTL之后之后的逻辑
3. 加入TTL之后的逻辑如下：

    public String test(String userName){

        // 加锁
        Boolean result = Optional.ofNullable(stringRedisTemplate.opsForValue().setIfAbsent(userName, String.valueOf(3),10,TimeUnit.SECONDS)).orElse(Boolean.FALSE);

        if (!result){
            // 执行加锁失败的逻辑
            return null;
        }

        try {
            //执行业务逻辑
        }catch (Exception e){

        }finally {
            // 释放锁
            stringRedisTemplate.delete(userName);
        }

        return null;
    }

在加入TTL之后呢，虽然解决了死锁的情况但是又引入如下的几个问题：
a.过期时间设置多久合适呢？可能每个请求执行的复杂度都不一样耗时自然也是不一样的这个没有办法去评估
b. 由于过期时间设置得不合理导致第一个线程逻辑还没有执行完成第二个线程已经加锁成功了，但是在释放释放锁的时候第一个线程把第二个线程加的锁释放了。这样导致锁的失效。那这个还不简单我给每个请求搞一个标识呗。
4. 加了标识之后的分布式锁的实现逻辑如下：

    public String test(String userName){
        
        // 身份标识
        String uuid = UUID.randomUUID().toString();
        // 加锁
        Boolean result = Optional.ofNullable(stringRedisTemplate.opsForValue().setIfAbsent(userName, uuid,10,TimeUnit.SECONDS)).orElse(Boolean.FALSE);

        if (!result){
            // 执行加锁失败的逻辑
            return null;
        }

        try {
            //执行业务逻辑
        }catch (Exception e){

        }finally {
            // 释放锁
            if (Objects.equals(stringRedisTemplate.opsForValue().get(userName), uuid)){
                stringRedisTemplate.delete(userName);
            }
           
        }

        return null;
    }

上面的这种实现能解决问题吗？看似解决了其实并没有解决。我们看看在下面这种情况下回出现啥问题：

从上面的图中可以发现线程1把线程2设置key删除了导致锁失效了。其实出现这个问题的原因如下：
a.比较uuid是否相同然后在删除key这个操作不是原子会被中断
b.还是过期时间的问题，一个线程所有逻辑还没有被执行玩key就已经被过期。这个问题出现导致第一个问题的出现。那要解决这个问题是不是只需要这样做在一个线程里面只要逻辑没有没有完成就搞一个定时任务去不停的刷新过期时间。当逻辑执行完成之后才去删除key这样是不是就没有问题了呀。其实redission就是这样做的。
5. 总结一下实现一个分布式锁需要那些注意事项
a. 不能出现死锁可以通过ttl去实现、ttl的续约可以使用一个单独的线程来做
b.一个线程加的锁不能被其他线程释放这个可以通过执行者的唯一标识来实现

1. 在看源码之前先看看Redission分布式锁的使用

    public String test(String userName){
        //获取锁对象
        RLock lock = redisson.getLock(userName);
        //加分布式锁
        lock.lock();

        try {
            //执行业务逻辑
        }catch (Exception e){

        }finally {
            lock.unlock();
        }

        return null;
    }

使用起来非常简单就和java里面基于AQS实现的互斥锁一样。接下来看看内部实现原理
2. 源码分析

// 加锁的核心逻辑,在使用lock.lock()时 leaseTime=-1 unit=null
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
        // 当前请求线程Id
        long threadId = Thread.currentThread().getId();
        // 核心加锁逻辑,请看tryAcquire的核心逻辑
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return;
        }

        RFuture future = subscribe(threadId);
        commandExecutor.syncSubscription(future);

        try {
            while (true) {
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }

                // waiting for message
                if (ttl >= 0) {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    getEntry(threadId).getLatch().acquire();
                }
            }
        } finally {
            unsubscribe(future, threadId);
        }
//        get(lockAsync(leaseTime, unit));
    }
    // 加锁的核心逻辑 请求参数 leaseTime=-1 leaseTime=null threadId当前请求线程Id
    private Long tryAcquire(long leaseTime, TimeUnit leaseTime, long threadId) {
        return get(tryAcquireAsync(leaseTime, unit, threadId));
    }
// 异步获取锁开始加锁
private  RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
        if (leaseTime != -1) {
            return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }
        RFuture ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        ttlRemainingFuture.addListener(new FutureListener() {
            @Override
            public void operationComplete(Future future) throws Exception {
                if (!future.isSuccess()) {
                    return;
                }

                Long ttlRemaining = future.getNow();
                // lock acquired
                if (ttlRemaining == null) {
                   // 执行成功之后开始走这段逻辑
                    scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;
    }
// 具体的加锁逻辑
// 请求参数 leaseTime=-1 unit=null threadId当前请求线程Id command redis中的EVAL命令
 RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                      "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "return redis.call('pttl', KEYS[1]);",
                    Collections.