Redis分布式鎖真的安全嗎？

今天我們來聊一聊Redis分布式鎖。

首先大家可以先思考一個簡單的問題，為什么要使用分布式鎖？普通的jvm鎖為什么不可以？

這個時候，大家肯定會吧啦吧啦想到一堆，例如java應(yīng)用屬于進程級，不同的ecs中部署相同的應(yīng)用，他們之間相互獨立。

所以，在分布式系統(tǒng)中，當(dāng)有多個客戶端需要獲取鎖時，我們需要分布式鎖。此時，鎖是保存在一個共享存儲系統(tǒng)中的，可以被多個客戶端共享訪問和獲取。

分布式鎖（SET NX）

知道了分布式鎖的使用場景，我們來自己簡單的實現(xiàn)下分布式鎖：

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);
}
}
}

以上代碼簡單的實現(xiàn)了一個扣減庫存的業(yè)務(wù)邏輯，我們拆開來說下都做了什么事情：

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

2、其次UUID.randomUUID().toString();生成該次請求的requestId，為什么需要生成這個唯一的UUID，后面在解鎖的時候會說到

3、獲取分布式鎖，通過stringRedisTemplate.opsForValue().setIfAbsent來實現(xiàn)，該語句的意思是如果存在該key則返回false，若不存在則進行key的設(shè)置，設(shè)置成功后返回true，將當(dāng)前線程獲取的uuid設(shè)置成value，給定一個鎖的過期時間，防止該線程無限制持久鎖導(dǎo)致死鎖，也為了防止該服務(wù)器突然宕機，導(dǎo)致其他機器的應(yīng)用無法獲取該鎖，這個是必須要做的設(shè)置，至于過期的時間，可以根據(jù)內(nèi)層業(yè)務(wù)邏輯的執(zhí)行時間來決定

4、執(zhí)行內(nèi)層的業(yè)務(wù)邏輯，進行扣庫存的操作

5、業(yè)務(wù)邏輯執(zhí)行完成后，走到finally的解鎖操作，進行解鎖操作時，首先我們來判斷當(dāng)前鎖的值是否為該線程持有的，防止當(dāng)前線程執(zhí)行較慢，導(dǎo)致鎖過期，從而刪除了其他線程持有的分布式鎖，對于該操作，我來舉個例子：

時刻1：線程A獲取分布式鎖，開始執(zhí)行業(yè)務(wù)邏輯

時刻2：線程B等待分布式鎖釋放

時刻3：線程A所在機器IO處理緩慢、GC pause等問題導(dǎo)致處理緩慢

時刻4：線程A依舊處于block狀態(tài)，鎖過期

時刻5：線程B獲取分布式鎖，開始執(zhí)行業(yè)務(wù)邏輯，此時線程A結(jié)束block，開始釋放鎖

時刻6：線程B處理業(yè)務(wù)邏輯緩慢，線程A釋放分布式鎖，但是此時釋放的是線程B的鎖，導(dǎo)致其他線程可以開始獲取鎖

看到這里，為什么每個請求需要requestId，并且在釋放鎖的情況下判斷是否是當(dāng)前的requestId是有必要的。

以上，就是一個簡單的分布式鎖的實現(xiàn)過程。但是你覺得上述實現(xiàn)還存在問題嗎？

答案是肯定的。若是在判斷完分布式鎖的value與requestId之后，鎖過期了，依然會存在以上問題。

那么有沒有什么辦法可以規(guī)避以上問題，讓我們不需要去完成這些實現(xiàn)，只需要專注于業(yè)務(wù)邏輯呢？

我們可以使用Redisson，并且Redisson有中文文檔，方便英文不好的同學(xué)查看（開發(fā)團隊中有中國的jackygurui）。

接下來我們再把上述代碼簡單的改造下就可以規(guī)避這些問題：

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()解決了這個問題，看到這里，是不是有同學(xué)會問，如果服務(wù)器宕機了，分布式鎖會一直存在嗎，也沒有去指定過期時間？

redisson分布式鎖中有一個watchdog機制，即會給一個leaseTime，默認為30s，到期后鎖自動釋放，如果一直沒有解鎖，watchdog機制會一直重新設(shè)定鎖的過期時間，通過設(shè)置TimeTask，延遲10s再次執(zhí)行鎖續(xù)命，將鎖的過期時間重置為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.