Redis分布式锁真的安全吗？

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

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

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

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

分布式锁（SET NX）

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

publicclassIndexController{

publicStringdeductStock(){

StringlockKey="lock:product_101";

//setNx获取分布式锁
StringclientId=UUID.randomUUID().toString();
Booleanresult=stringRedisTemplate.opsForValue().setIfAbsent(lockKey,clientId,30,TimeUnit.SECONDS);//jedis.setnx(k,v)
if(!result){
return"error_code";
}
try{
intstock=Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//jedis.get("stock")
if(stock>0){
intrealStock=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）。

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

publicclassIndexController{

publicStringdeductStock(){

StringlockKey="lock:product_101";

//setNx获取分布式锁
//StringclientId=UUID.randomUUID().toString();
//Booleanresult=stringRedisTemplate.opsForValue().setIfAbsent(lockKey,clientId,30,TimeUnit.SECONDS);//jedis.setnx(k,v)
//获取锁对象
RLockredissonLock=redisson.getLock(lockKey);
//加分布式锁
redissonLock.lock();
try{
intstock=Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//jedis.get("stock")
if(stock>0){
intrealStock=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的最终加锁方法：

RFuturetryLockInnerAsync(longleaseTime,TimeUnitunit,longthreadId,RedisStrictCommandcommand){
internalLockLeaseTime=unit.toMillis(leaseTime);

returncommandExecutor.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]);"+
"returnnil;"+
"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]);"+
"returnnil;"+
"end;"+
"returnredis.call('pttl',KEYS[1]);",
Collections.