Redis分布式鎖的使用場(chǎng)景

分布式鎖（SET NX）

分布式鎖Redlock

基于ZooKeeper的分布式鎖更安全嗎？

總結(jié)

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

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

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

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

分布式鎖（SET NX）

知道了分布式鎖的使用場(chǎng)景，我們來自己簡(jiǎn)單的實(shí)現(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("扣減成功，剩余庫(kù)存:"+realStock);
}else{
System.out.println("扣減失敗，庫(kù)存不足");
}
}finally{
//解鎖
if(clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))){
stringRedisTemplate.delete(lockKey);
}
}
}

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

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

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

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

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

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

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

時(shí)刻2：線程B等待分布式鎖釋放

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

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

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

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

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

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

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

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

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

接下來我們?cè)侔焉鲜龃a簡(jiǎn)單的改造下就可以規(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)
//獲取鎖對(duì)象
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("扣減成功，剩余庫(kù)存:"+realStock);
}else{
System.out.println("扣減失敗，庫(kù)存不足");
}
}finally{
//解鎖
//if(clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))){
//stringRedisTemplate.delete(lockKey);
//}
//redisson分布式鎖解鎖
redissonLock.unlock();
}
}

可以看到，使用redisson分布式鎖會(huì)簡(jiǎn)單很多，我們通過redissonLock.lock()和redissonLock.unlock()解決了這個(gè)問題，看到這里，是不是有同學(xué)會(huì)問，如果服務(wù)器宕機(jī)了，分布式鎖會(huì)一直存在嗎，也沒有去指定過期時(shí)間？

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