Java手寫分布式鎖的實現(xiàn)

Part1 前言

隨著互聯(lián)網(wǎng)業(yè)務的發(fā)展，原本單機部署的系統(tǒng)演化成如今的分布式集群系統(tǒng)后，由于分布式系統(tǒng)多線程，多進程并且分布在不同的機器上，這會使原本的單機鎖失效，而且單純的Java API并不能提供分布式鎖的能力，為了解決這個問題就需要一種跨JVM的互斥機制來控制共享資源的訪問，這就是分布式鎖要解決的問題。

本文將從實現(xiàn)本地鎖所產(chǎn)生的問題入手，從而介紹分布式鎖主流的實現(xiàn)方案，重點實現(xiàn)基于Redis的分布式鎖。

本地鎖會出現(xiàn)的問題（此篇幅代碼圖片過多，因此放在最后）

Part2 分布式鎖的實現(xiàn)

1 分布式鎖主要的實現(xiàn)有：

基于數(shù)據(jù)庫實現(xiàn)分布式鎖

基于緩存（Redis等,本文基于Redis實現(xiàn)手寫分布式鎖 ，因為這樣可以更好的理解分布式鎖的原理及實現(xiàn)，當然也可以使用Redisson）

基于Zookeeper

2 每種分布式鎖的解決方案都有各自的優(yōu)缺點

性能角度：redis > zk > mysql

安全角度：zk > redis == mysql

難易程度：zk > redis > mysql

3 分布式鎖要具備的特點：

獨占排他互斥

可以通過 setnx （redis命令：執(zhí)行多次，只有一次能夠成功）

set key value ex 3 nx

防死鎖發(fā)生

請求獲取到鎖之后，服務器掛掉了，導致鎖無法釋放：給lock鎖添加過期時間

可以通過redis命令 expire

或者通過 set key value ex 3 nx

保證原子性

redis是單線程的，接受或者執(zhí)行指令遵循one-by-one原則。只要指令之間不被插入其他指令即可保證原子性，lua腳本批量發(fā)送多個指令給redis服務器，lua腳本也可以實現(xiàn)一些業(yè)務邏輯，redis集成了lua腳本，可以直接使用eval指令執(zhí)行l(wèi)ua腳本。

獲取鎖和設置過期時間之間

判斷和刪除之間：lua腳本

防誤刪：

uuid給每個線程的鎖添加唯一標識

自動續(xù)期

可重入：hash數(shù)據(jù)結(jié)構(gòu) + lua腳本

集群情況下，可能導致鎖失效：RedLock算法（redis特有的）

一個請求從主中獲取到鎖，從還沒來得及同步數(shù)據(jù)，主就掛掉了，從就升級為新主，新的請求就可以從新主中獲取鎖

4 基于redis分布式鎖的基本實現(xiàn)

我們可以借助于redis中的命令setnx(key, value)，key不存在就新增，存在就什么都不做。同時有多個客戶端發(fā)送setnx命令，只有一個客戶端可以成功，返回1（true）；其他的客戶端返回0（false），流程圖如下圖所示：

多個客戶端同時嘗試獲取鎖（setnx）

獲取成功，執(zhí)行業(yè)務邏輯，執(zhí)行完成釋放鎖（del）

其他客戶端等待重試

代碼實現(xiàn)：

publicvoidtestLock(){
//1.從redis中獲取鎖,setnx
Booleanlock=this.redisTemplate.opsForValue().setIfAbsent("lock","111");
if(lock){
//查詢redis中的num值
Stringvalue=this.redisTemplate.opsForValue().get("num");
//沒有該值return
if(StringUtils.isBlank(value)){
return;
}
//有值就轉(zhuǎn)成成int
intnum=Integer.parseInt(value);
//把redis中的num值+1
this.redisTemplate.opsForValue().set("num",String.valueOf(++num));
//2.釋放鎖del
this.redisTemplate.delete("lock");
}else{
//3.每隔1秒鐘回調(diào)一次，再次嘗試獲取鎖
try{
Thread.sleep(1000);
testLock();
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}

那么以上代碼是否可以解決全部問題呢？ 顯示是不能的，我們假設setnx剛好獲取到鎖，業(yè)務邏輯出現(xiàn)異常，導致鎖無法釋放，怎么辦呢？

5 優(yōu)化分布式鎖_設置過期時間

設置過期有倆種方式可以選擇：

通過expire設置過期時間（缺乏原子性：如果在setnx和expire之間出現(xiàn)異常，鎖也無法釋放）

在set時指定過期時間（推薦）

代碼實現(xiàn)優(yōu)化就是在設置鎖的時候設置過期時間:

publicvoidtestLock(){
//1.從redis中獲取鎖,setnx
Booleanlock=this.redisTemplate.opsForValue().setIfAbsent("lock","111",3,TimeUnit.MINUTES);
if(lock){
//與之前相同代碼略過
...
}
}

那么還會不會存在問題呢?

場景：如果業(yè)務邏輯的執(zhí)行時間是7s。執(zhí)行流程如下:

index1業(yè)務邏輯沒執(zhí)行完，3秒后鎖被自動釋放。

index2獲取到鎖，執(zhí)行業(yè)務邏輯，3秒后鎖被自動釋放。

index3獲取到鎖，執(zhí)行業(yè)務邏輯

index1業(yè)務邏輯執(zhí)行完成，開始調(diào)用del釋放鎖，這時釋放的是index3的鎖，導致index3的業(yè)務只執(zhí)行1s就被別人釋放。

最終等于沒鎖的情況。

解決：setnx獲取鎖時，設置一個指定的唯一值（例如：uuid）；釋放前獲取這個值，判斷是否自己的鎖

6 優(yōu)化分布式鎖_防止誤刪除

publicvoidtestLock(){
//1.從redis中獲取鎖,setnx
Stringuuid=UUID.randomUUID().toString();
Booleanlock=this.redisTemplate.opsForValue().setIfAbsent("lock",uuid,3,TimeUnit.MINUTES);
if(lock){
//與之前相同代碼略過
...
//2.釋放鎖del
if(StringUtils.equals(redisTemplate.opsForValue().get("lock"),uuid)){
this.redisTemplate.delete("lock");
}
}
}

場景：

index1執(zhí)行刪除時，查詢到的lock值確實和uuid相等

index1執(zhí)行刪除前，lock剛好過期時間已到，被redis自動釋放

index2獲取了lock

index1執(zhí)行刪除，此時會把index2的lock刪除

問題：缺乏原子性

7 優(yōu)化分布式鎖_LUA腳本保證刪除的原子性

首先我們先簡單介紹一下lua腳本的基本知識（lua腳本是c語言）

定義變量：

全局變量：a = 11

局部變量：local b = 22

redis不允許lua腳本創(chuàng)建全局變量，只能聲明局部變量

流程控制:

if(exp)then
業(yè)務邏輯
elseif(exp)then
業(yè)務邏輯
else
業(yè)務邏輯
end

redis中執(zhí)行l(wèi)ua腳本:

eval script numkeys keys[] args[] ： eval指令的輸出不是lua腳本的打印而是lua腳本的返回值

script：lua腳本字符串，定義動態(tài)變量：KEYS[1] ARGV[1]

numkeys：key數(shù)組的元素個數(shù)

keys：keys數(shù)組

args：argv數(shù)組

redis集群執(zhí)行l(wèi)ua腳本可能會報錯：如果所有keys不在同一個分片上，lua腳本就會報錯：解決方案是:

keys只傳一個

可以使用CLUSTER KEYSLOT bb{xx}

刪除LUA腳本：

ifredis.call('get',KEYS[1])==ARGV[1]thenreturnredis.call('del',KEYS[1])elsereturn0end
publicvoidtestLock(){
//1.從redis中獲取鎖,setnx
Stringuuid=UUID.randomUUID().toString();
Booleanlock=this.redisTemplate.opsForValue().setIfAbsent("lock",uuid,3,TimeUnit.SECONDS);
if(lock){
//與之前相同代碼略過
...
//2.釋放鎖del
Stringscript="ifredis.call('get',KEYS[1])==ARGV[1]thenreturnredis.call('del',KEYS[1])elsereturn0end";
this.redisTemplate.execute(newDefaultRedisScript<>(script),Arrays.asList("lock"),uuid);
}else{
//3.每隔1秒鐘回調(diào)一次，再次嘗試獲取鎖
try{
Thread.sleep(1000);
testLock();
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}

8 優(yōu)化分布式鎖_可以重入

上述加鎖命令使用了 SETNX ，一旦鍵存在就無法再設置成功，這就導致后續(xù)同一線程內(nèi)繼續(xù)加鎖，將會加鎖失敗。當一個線程執(zhí)行一段代碼成功獲取鎖之后，繼續(xù)執(zhí)行時，又遇到加鎖的子任務代碼，可重入性就保證線程能繼續(xù)執(zhí)行，而不可重入就是需要等待鎖釋放之后，再次獲取鎖成功，才能繼續(xù)往下執(zhí)行。

可重入鎖最大特性就是計數(shù)，計算加鎖的次數(shù)。所以當可重入鎖需要在分布式環(huán)境實現(xiàn)時，我們也就需要統(tǒng)計加鎖次數(shù)。

我們基于Redis Hash 實現(xiàn)方案 ：

Redis 提供了 Hash （哈希表）這種可以存儲鍵值對數(shù)據(jù)結(jié)構(gòu)。所以我們可以使用 Redis Hash 存儲的鎖的重入次數(shù)，然后利用 lua 腳本判斷邏輯。

加鎖

if(redis.call('exists',KEYS[1])==0orredis.call('hexists',KEYS[1],ARGV[1])==1)
then
redis.call('hincrby',KEYS[1],ARGV[1],1);
redis.call('expire',KEYS[1],ARGV[2]);
return1;
else
return0;
end

假設值為：KEYS:[lock], ARGV[uuid, expire]

如果鎖不存在或者這是自己的鎖，就通過hincrby（不存在新增，存在就加1）獲取鎖或者鎖次數(shù)加1。 代碼實例如下：

privateBooleantryLock(StringlockName,Stringuuid,Longexpire){
Stringscript="if(redis.call('exists',KEYS[1])==0orredis.call('hexists',KEYS[1],ARGV[1])==1)"+
"then"+
"redis.call('hincrby',KEYS[1],ARGV[1],1);"+
"redis.call('expire',KEYS[1],ARGV[2]);"+
"return1;"+
"else"+
"return0;"+
"end";
if(!this.redisTemplate.execute(newDefaultRedisScript<>(script,Boolean.class),Arrays.asList(lockName),uuid,expire.toString())){
try{
//沒有獲取到鎖，重試
Thread.sleep(200);
tryLock(lockName,uuid,expire);
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
//獲取到鎖，返回true
returntrue;
}

解鎖

--判斷hashset可重入key的值是否等于0
--如果為nil代表自己的鎖已不存在，在嘗試解其他線程的鎖，解鎖失敗
--如果為0代表可重入次數(shù)被減1
--如果為1代表該可重入key解鎖成功
if(redis.call('hexists',KEYS[1],ARGV[1])==0)then
returnnil;
end;
--小于等于0代表可以解鎖
if(redis.call('hincrby',KEYS[1],ARGV[1],-1)>0)then
return0;
else
redis.call('del',KEYS[1]);
return1;
end;

這里之所以沒有跟加鎖一樣使用 Boolean ,這是因為解鎖 lua 腳本中，三個返回值含義如下：

1 代表解鎖成功，鎖被釋放

0 代表可重入次數(shù)被減 1

null 代表其他線程嘗試解鎖，解鎖失敗

如果返回值使用 Boolean，Spring-data-redis 進行類型轉(zhuǎn)換時將會把 null 轉(zhuǎn)為 false，這就會影響我們邏輯判斷，所以返回類型只好使用 Long。

privatevoidunlock(StringlockName,Stringuuid){
Stringscript="if(redis.call('hexists',KEYS[1],ARGV[1])==0)then"+
"returnnil;"+
"end;"+
"if(redis.call('hincrby',KEYS[1],ARGV[1],-1)>0)then"+
"return0;"+
"else"+
"redis.call('del',KEYS[1]);"+
"return1;"+
"end;";
//這里之所以沒有跟加鎖一樣使用Boolean,這是因為解鎖lua腳本中，三個返回值含義如下：
//1代表解鎖成功，鎖被釋放
//0代表可重入次數(shù)被減1
//null代表其他線程嘗試解鎖，解鎖失敗
Longresult=this.redisTemplate.execute(newDefaultRedisScript<>(script,Long.class),Lists.newArrayList(lockName),uuid);
//如果未返回值，代表嘗試解其他線程的鎖
if(result==null){
thrownewIllegalMonitorStateException("attempttounlocklock,notlockedbylockName:"
+lockName+"withrequest:"+uuid);
}
}

使用

publicvoidtestLock(){
//加鎖
Stringuuid=UUID.randomUUID().toString();
Booleanlock=this.tryLock("lock",uuid,300l);
if(lock){
//讀取redis中的num值
StringnumString=this.redisTemplate.opsForValue().get("num");
if(StringUtils.isBlank(numString)){
return;
}
//++操作
Integernum=Integer.parseInt(numString);
num++;
//放入redis
this.redisTemplate.opsForValue().set("num",String.valueOf(num));
//測試可重入性
this.testSubLock(uuid);
//釋放鎖
this.unlock("lock",uuid);
}
}
//測試可重入性
privatevoidtestSubLock(Stringuuid){
//加鎖
Booleanlock=this.tryLock("lock",uuid,300l);
if(lock){
System.out.println("分布式可重入鎖。。。");
this.unlock("lock",uuid);
}
}

9 優(yōu)化分布式鎖_自動續(xù)期

A線程超時時間設為10s（為了解決死鎖問題），但代碼執(zhí)行時間可能需要30s，然后redis服務端10s后將鎖刪除。 此時，B線程恰好申請鎖，redis服務端不存在該鎖，可以申請，也執(zhí)行了代碼。

那么問題來了， A、B線程都同時獲取到鎖并執(zhí)行業(yè)務邏輯，這與分布式鎖最基本的性質(zhì)相違背：在任意一個時刻，只有一個客戶端持有鎖（即獨享排他）。

鎖延期方法：開啟子線程執(zhí)行延期

/**
*鎖延期
*線程等待超時時間的2/3時間后,執(zhí)行鎖延時代碼,直到業(yè)務邏輯執(zhí)行完畢,因此在此過程中,其他線程無法獲取到鎖,保證了線程安全性
*@paramlockName
*@paramexpire單位：毫秒
*/
privatevoidrenewTime(StringlockName,Stringuuid,Longexpire){
Stringscript="if(redis.call('hexists',KEYS[1],ARGV[1])==1)thenredis.call('expire',KEYS[1],ARGV[2]);return1;elsereturn0;end";
newThread(()->{
while(this.redisTemplate.execute(newDefaultRedisScript<>(script,Boolean.class),Lists.newArrayList(lockName),uuid,expire.toString())){
try{
//到達過期時間的2/3時間，自動續(xù)期
Thread.sleep(expire/3);
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}).start();
}

獲取鎖成功后，調(diào)用延期方法給鎖 定時延期：

privateBooleantryLock(StringlockName,Stringuuid,Longexpire){
Stringscript="if(redis.call('exists',KEYS[1])==0orredis.call('hexists',KEYS[1],ARGV[1])==1)"+
"then"+
"redis.call('hincrby',KEYS[1],ARGV[1],1);"+
"redis.call('expire',KEYS[1],ARGV[2]);"+
"return1;"+
"else"+
"return0;"+
"end";
if(!this.redisTemplate.execute(newDefaultRedisScript<>(script,Boolean.class),Arrays.asList(lockName),uuid,expire.toString())){
try{
//沒有獲取到鎖，重試
Thread.sleep(200);
tryLock(lockName,uuid,expire);
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
//鎖續(xù)期
this.renewTime(lockName,uuid,expire*1000);
//獲取到鎖，返回true
returntrue;
}

10 優(yōu)化分布式鎖_Redlock算法

redis集群狀態(tài)下的問題：

客戶端A從master獲取到鎖

在master將鎖同步到slave之前，master宕掉了。

slave節(jié)點被晉級為master節(jié)點

客戶端B取得了同一個資源被客戶端A已經(jīng)獲取到的另外一個鎖。安全失效 解決集群下鎖失效，參照redis官方網(wǎng)站針對redlock文檔：redis.io/topics/dist… [1]

11 本地鎖會出現(xiàn)的問題

我們知道java中有synchronized、lock鎖、讀寫鎖ReadWriteLock，眾所周知這些鎖都是本地鎖。

提到鎖就不得不提JUC:java.util.concurrent包，又稱concurrent包。jdk1.5提供，為多線程高并發(fā)編程而提供的包,但此文章的場景是分布式場景，后續(xù)會出JUC的文章。

簡單的介紹一下synchronized及l(fā)ock鎖

synchronized是一個關鍵字，lock是一個接口，ReentrantLock是實現(xiàn)了lock接口的一個類

ReentrantLock：悲觀的獨占的互斥的排他的可公平可不公平的可重入鎖

synchronized：悲觀的獨占的互斥的排他的非公平的可重入鎖

準備

redis、ab工具(壓測)

不使用任何鎖的情況下

我們首先創(chuàng)建一個測試方法，testNoLock

@GetMapping("/test")
publicvoidtestNoLock(){
Stringcount=(String)this.redisTemplate.opsForValue().get("count");
if(count==null){
//沒有值直接返回
return;
}
//有值就轉(zhuǎn)成成int
intnumber=Integer.parseInt(count);
//把redis中的num值+1
this.redisTemplate.opsForValue().set("count",String.valueOf(++number));
}

測試之前的查看值為1

@GetMapping("/getCount")
publicStringgetCount(){
Stringcount=String.valueOf(this.redisTemplate.opsForValue().get("count"));
returncount;//1
}

接下來使用ab壓力測試工具

//ab-n（一次發(fā)送的請求數(shù)）-c（請求的并發(fā)數(shù)）訪問路徑
ab-n100-c50http://127.0.0.1:8080/test/test

再次查詢結(jié)果為6，說明問題很大

使用本地鎖

publicsynchronizedvoidtestNoLock(){
Stringcount=String.valueOf(this.redisTemplate.opsForValue().get("count"));
if("null".equals(count)){
//沒有值直接返回
return;
}
//有值就轉(zhuǎn)成成int
intnumber=Integer.parseInt(count);
//把redis中的num值+1
this.redisTemplate.opsForValue().set("count",String.valueOf(++number));
}

再次使用ab壓力測試工具

ab-n100-c50http://127.0.0.1:8080/test/test

此次結(jié)果為106，說明結(jié)果是正確的，看樣子結(jié)果是非常完美的，但是真的很完美嗎？

使用集群+本地鎖

我們只需要在idea中在啟動倆個服務，修改端口號，三個運行實例的名稱是相同的，并且網(wǎng)關的配置就是通過服務名在負載均衡，所以我們只需要訪問網(wǎng)關，網(wǎng)關就會給我們做負載均衡了。

再次使用ab壓力測試工具（將count重置為1）

ab-n100-c50http://127.0.0.1:8080/test/test

此次的結(jié)果為58?。?！

到此我們可以知道，本地鎖是有局限性的。

審核編輯：劉清