Redis分布式锁真的安全吗？

今天我们来聊一聊Redis分布式锁。

首先大家可以先思考一个简单的问题，为什么要使用分布式锁？普通的jvm锁为什么不可以？

这个时候，大家肯定会吧啦吧啦想到一堆，例如java应用属于进程级，不同的ecs中部署相同的应用，他们之间相互独立。

所以，在分布式系统中，当有多个客户端需要获取锁时，我们需要分布式锁。此时，锁是保存在一个共享存储系统中的，可以被多个客户端共享访问和获取。

分布式锁（SET NX）

知道了分布式锁的使用场景，我们来自己简单的实现下分布式锁：

 

public class IndexController {

    public String deductStock() {

        String lockKey = "lock:product_101";

        //setNx 获取分布式锁
        String clientId = UUID.randomUUID().toString();
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS); //jedis.setnx(k,v)
        if (!result) {
            return "error_code";
        }
        try {
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
                System.out.println("扣减成功，剩余库存:" + realStock);
            } else {
                System.out.println("扣减失败，库存不足");
            }
        } finally {
            //解锁
            if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
                stringRedisTemplate.delete(lockKey);
            }
    }
}

 

以上代码简单的实现了一个扣减库存的业务逻辑，我们拆开来说下都做了什么事情：

1、首先声明了lockkey，表示我们需要set的keyName

2、其次UUID.randomUUID().toString();生成该次请求的requestId，为什么需要生成这个唯一的UUID，后面在解锁的时候会说到

3、获取分布式锁，通过stringRedisTemplate.opsForValue().setIfAbsent来实现，该语句的意思是如果存在该key则返回false，若不存在则进行key的设置，设置成功后返回true，将当前线程获取的uuid设置成value，给定一个锁的过期时间，防止该线程无限制持久锁导致死锁，也为了防止该服务器突然宕机，导致其他机器的应用无法获取该锁，这个是必须要做的设置，至于过期的时间，可以根据内层业务逻辑的执行时间来决定

4、执行内层的业务逻辑，进行扣库存的操作

5、业务逻辑执行完成后，走到finally的解锁操作，进行解锁操作时，首先我们来判断当前锁的值是否为该线程持有的，防止当前线程执行较慢，导致锁过期，从而删除了其他线程持有的分布式锁，对于该操作，我来举个例子：

时刻1：线程A获取分布式锁，开始执行业务逻辑

时刻2：线程B等待分布式锁释放

时刻3：线程A所在机器IO处理缓慢、GC pause等问题导致处理缓慢

时刻4：线程A依旧处于block状态，锁过期

时刻5：线程B获取分布式锁，开始执行业务逻辑，此时线程A结束block，开始释放锁

时刻6：线程B处理业务逻辑缓慢，线程A释放分布式锁，但是此时释放的是线程B的锁，导致其他线程可以开始获取锁

看到这里，为什么每个请求需要requestId，并且在释放锁的情况下判断是否是当前的requestId是有必要的。

以上，就是一个简单的分布式锁的实现过程。但是你觉得上述实现还存在问题吗？

答案是肯定的。若是在判断完分布式锁的value与requestId之后，锁过期了，依然会存在以上问题。

那么有没有什么办法可以规避以上问题，让我们不需要去完成这些实现，只需要专注于业务逻辑呢？

我们可以使用Redisson，并且Redisson有中文文档，方便英文不好的同学查看（开发团队中有中国的jackygurui）。

接下来我们再把上述代码简单的改造下就可以规避这些问题：

 

public class IndexController {

    public String deductStock() {

        String lockKey = "lock:product_101";

        //setNx 获取分布式锁
        //String clientId = UUID.randomUUID().toString();
        //Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS); //jedis.setnx(k,v)
        //获取锁对象
        RLock redissonLock = redisson.getLock(lockKey);
        //加分布式锁
        redissonLock.lock();
        try {
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
                System.out.println("扣减成功，剩余库存:" + realStock);
            } else {
                System.out.println("扣减失败，库存不足");
            }
        } finally {
            //解锁
            //if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
            //    stringRedisTemplate.delete(lockKey);
            //}
            //redisson分布式锁解锁
            redissonLock.unlock();
    }
}

 

可以看到，使用redisson分布式锁会简单很多，我们通过redissonLock.lock()和redissonLock.unlock()解决了这个问题，看到这里，是不是有同学会问，如果服务器宕机了，分布式锁会一直存在吗，也没有去指定过期时间？

redisson分布式锁中有一个watchdog机制，即会给一个leaseTime，默认为30s，到期后锁自动释放，如果一直没有解锁，watchdog机制会一直重新设定锁的过期时间，通过设置TimeTask，延迟10s再次执行锁续命，将锁的过期时间重置为30s。下面就从redisson.lock()的源码来看下：

lock的最终加锁方法：

 

 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.