Part1 前言

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

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

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

Part2 分布式鎖的實現

1 分布式鎖主要的實現有：

基于數據庫實現分布式鎖

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

基于Zookeeper

2 每種分布式鎖的解決方案都有各自的優缺點

性能角度：redis > zk > mysql

安全角度：zk > redis == mysql

難易程度：zk > redis > mysql

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

獨占排他互斥

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

set key value ex 3 nx

防死鎖發生

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

可以通過redis命令 expire

或者通過 set key value ex 3 nx

保證原子性

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

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

判斷和刪除之間：lua腳本

防誤刪：

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

自動續期

可重入：hash數據結構 + lua腳本

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

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

4 基于redis分布式鎖的基本實現

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

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

獲取成功，執行業務邏輯，執行完成釋放鎖（del）

其他客戶端等待重試

代碼實現：

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;
}
//有值就轉成成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秒鐘回調一次，再次嘗試獲取鎖
try{
Thread.sleep(1000);
testLock();
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}

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

5 優化分布式鎖_設置過期時間

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

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

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

代碼實現優化就是在設置鎖的時候設置過期時間:

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

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

場景：如果業務邏輯的執行時間是7s。執行流程如下:

index1業務邏輯沒執行完，3秒后鎖被自動釋放。

index2獲取到鎖，執行業務邏輯，3秒后鎖被自動釋放。

index3獲取到鎖，執行業務邏輯

index1業務邏輯執行完成，開始調用del釋放鎖，這時釋放的是index3的鎖，導致index3的業務只執行1s就被別人釋放。

最終等于沒鎖的情況。

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

6 優化分布式鎖_防止誤刪除

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執行刪除時，查詢到的lock值確實和uuid相等

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

index2獲取了lock

index1執行刪除，此時會把index2的lock刪除

問題：缺乏原子性

7 優化分布式鎖_LUA腳本保證刪除的原子性

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

定義變量：

全局變量：a = 11

局部變量：local b = 22

redis不允許lua腳本創建全局變量，只能聲明局部變量

流程控制:

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

redis中執行lua腳本:

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

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

numkeys：key數組的元素個數

keys：keys數組

args：argv數組

redis集群執行lua腳本可能會報錯：如果所有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秒鐘回調一次，再次嘗試獲取鎖
try{
Thread.sleep(1000);
testLock();
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}

8 優化分布式鎖_可以重入

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

可重入鎖最大特性就是計數，計算加鎖的次數。所以當可重入鎖需要在分布式環境實現時，我們也就需要統計加鎖次數。

我們基于Redis Hash 實現方案 ：

Redis 提供了 Hash （哈希表）這種可以存儲鍵值對數據結構。所以我們可以使用 Redis Hash 存儲的鎖的重入次數，然后利用 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）獲取鎖或者鎖次數加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代表可重入次數被減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 代表可重入次數被減 1

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

如果返回值使用 Boolean，Spring-data-redis 進行類型轉換時將會把 null 轉為 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代表可重入次數被減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 優化分布式鎖_自動續期

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

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

鎖延期方法：開啟子線程執行延期

/**
*鎖延期
*線程等待超時時間的2/3時間后,執行鎖延時代碼,直到業務邏輯執行完畢,因此在此過程中,其他線程無法獲取到鎖,保證了線程安全性
*@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時間，自動續期
Thread.sleep(expire/3);
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}).start();
}

獲取鎖成功后，調用延期方法給鎖 定時延期：

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();
}
}
//鎖續期
this.renewTime(lockName,uuid,expire*1000);
//獲取到鎖，返回true
returntrue;
}

10 優化分布式鎖_Redlock算法

redis集群狀態下的問題：

客戶端A從master獲取到鎖

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

slave節點被晉級為master節點

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

11 本地鎖會出現的問題

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

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

簡單的介紹一下synchronized及lock鎖

synchronized是一個關鍵字，lock是一個接口，ReentrantLock是實現了lock接口的一個類

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

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

準備

redis、ab工具(壓測)

不使用任何鎖的情況下

我們首先創建一個測試方法，testNoLock

@GetMapping("/test")
publicvoidtestNoLock(){
Stringcount=(String)this.redisTemplate.opsForValue().get("count");
if(count==null){
//沒有值直接返回
return;
}
//有值就轉成成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（一次發送的請求數）-c（請求的并發數）訪問路徑
ab-n100-c50http://127.0.0.1:8080/test/test

再次查詢結果為6，說明問題很大

使用本地鎖

publicsynchronizedvoidtestNoLock(){
Stringcount=String.valueOf(this.redisTemplate.opsForValue().get("count"));
if("null".equals(count)){
//沒有值直接返回
return;
}
//有值就轉成成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

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

使用集群+本地鎖

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

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

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

此次的結果為58！！！

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

審核編輯：劉清