if?(jedis.setnx(lockKey,?val)?==?1)?{
???jedis.expire(lockKey,?timeout);
}

String?result?=?jedis.set(lockKey,?requestId,?"NX",?"PX",?expireTime);
if?("OK".equals(result))?{
????return?true;
}
return?false;

try{
??String?result?=?jedis.set(lockKey,?requestId,?"NX",?"PX",?expireTime);
??if?("OK".equals(result))?{
??????return?true;
??}
??return?false;
}?finally?{
????unlock(lockKey);
}?

需要捕獲業(yè)務代碼的異常，然后在 finally 中釋放鎖。換句話說就是：無論代碼執(zhí)行成功或失敗了，都需要釋放鎖。

此時，有些朋友可能會問：假如剛好在釋放鎖的時候，系統(tǒng)被重啟了，或者網(wǎng)絡斷線了，或者機房斷點了，不也會導致釋放鎖失??？

這是一個好問題，因為這種小概率問題確實存在。但還記得前面我們給鎖設置過超時時間嗎？

即使出現(xiàn)異常情況造成釋放鎖失敗，但到了我們設定的超時時間，鎖還是會被 redis 自動釋放。但只在 finally 中釋放鎖，就夠了嗎？

做人要厚道，先回答上面的問題：只在 finally 中釋放鎖，當然是不夠的，因為釋放鎖的姿勢，還是不對。

哪里不對？

答：在多線程場景中，可能會出現(xiàn)釋放了別人的鎖的情況。

有些朋友可能會反駁：假設在多線程場景中，線程 A 獲取到了鎖，但如果線程 A 沒有釋放鎖，此時，線程 B 是獲取不到鎖的，何來釋放了別人鎖之說？

答：假如線程 A 和線程B，都使用 lockKey 加鎖。線程 A 加鎖成功了，但是由于業(yè)務功能耗時時間很長，超過了設置的超時時間。這時候，redis 會自動釋放 lockKey 鎖。

此時，線程 B 就能給 lockKey 加鎖成功了，接下來執(zhí)行它的業(yè)務操作。恰好這個時候，線程 A 執(zhí)行完了業(yè)務功能，接下來，在 finally 方法中釋放了鎖 lockKey。這不就出問題了，線程 B 的鎖，被線程 A 釋放了。

我想這個時候，線程 B 肯定哭暈在廁所里，并且嘴里還振振有詞。那么，如何解決這個問題呢？

不知道你們注意到?jīng)]？在使用 set 命令加鎖時，除了使用 lockKey 鎖標識，還多設置了一個參數(shù)：requestId，為什么要需要記錄 requestId 呢？

答：requestId 是在釋放鎖的時候用的。

if?(jedis.get(lockKey).equals(requestId))?{
????jedis.del(lockKey);
????return?true;
}
return?false;

在釋放鎖的時候，先獲取到該鎖的值（之前設置值就是 requestId），然后判斷跟之前設置的值是否相同，如果相同才允許刪除鎖，返回成功。如果不同，則直接返回失敗。

換句話說就是：自己只能釋放自己加的鎖，不允許釋放別人加的鎖。

這里為什么要用 requestId，用 userId 不行嗎？

答：如果用 userId 的話，對于請求來說并不唯一，多個不同的請求，可能使用同一個 userId。而 requestId 是全局唯一的，不存在加鎖和釋放鎖亂掉的情況。

if?redis.call('get',?KEYS[1])?==?ARGV[1]?then?
?return?redis.call('del',?KEYS[1])?
else?
??return?0?
end

lua 腳本能保證查詢鎖是否存在和刪除鎖是原子操作，用它來釋放鎖效果更好一些。

if?(redis.call('exists',?KEYS[1])?==?0)?then
????redis.call('hset',?KEYS[1],?ARGV[2],?1);?
????redis.call('pexpire',?KEYS[1],?ARGV[1]);?
?return?nil;?
end
if?(redis.call('hexists',?KEYS[1],?ARGV[2])?==?1)
???redis.call('hincrby',?KEYS[1],?ARGV[2],?1);?
???redis.call('pexpire',?KEYS[1],?ARGV[1]);?
??return?nil;?
end;?
return?redis.call('pttl',?KEYS[1]);

這是 redisson 框架的加鎖代碼，寫的不錯，大家可以借鑒一下。有趣，下面還有哪些好玩的東西？

上面的加鎖方法看起來好像沒有問題，但如果你仔細想想，如果有 1 萬的請求同時去競爭那把鎖，可能只有一個請求是成功的，其余的 9999 個請求都會失敗。

在秒殺場景下，會有什么問題？

答：每 1 萬個請求，有 1 個成功。再 1 萬個請求，有 1 個成功。如此下去，直到庫存不足。這就變成均勻分布的秒殺了，跟我們想象中的不一樣。

如何解決這個問題呢？

此外，還有一種場景：比如，有兩個線程同時上傳文件到 sftp，上傳文件前先要創(chuàng)建目錄。

假設兩個線程需要創(chuàng)建的目錄名都是當天的日期，比如：20210920，如果不做任何控制，直接并發(fā)的創(chuàng)建目錄，第二個線程必然會失敗。

這時候有些朋友可能會說：這還不容易，加一個 redis 分布式鎖就能解決問題了，此外再判斷一下，如果目錄已經(jīng)存在就不創(chuàng)建，只有目錄不存在才需要創(chuàng)建。

try?{
??String?result?=?jedis.set(lockKey,?requestId,?"NX",?"PX",?expireTime);
??if?("OK".equals(result))?{
????if(!exists(path))?{
???????mkdir(path);
????}
????return?true;
??}
}?finally{
????unlock(lockKey,requestId);
}??
return?false;

一切看似美好，但經(jīng)不起仔細推敲。來自靈魂的一問：第二個請求如果加鎖失敗了，接下來，是返回失敗，還是返回成功呢？

顯然第二個請求，肯定是不能返回失敗的，如果返回失敗了，這個問題還是沒有被解決。如果文件還沒有上傳成功，直接返回成功會有更大的問題。頭疼，到底該如何解決呢？

try?{
??Long?start?=?System.currentTimeMillis();
??while(true)?{
?????String?result?=?jedis.set(lockKey,?requestId,?"NX",?"PX",?expireTime);
?????if?("OK".equals(result))?{
????????if(!exists(path))?{
???????????mkdir(path);
????????}
????????return?true;
?????}

?????long?time?=?System.currentTimeMillis()?-?start;
??????if?(time>=timeout)?{
??????????return?false;
??????}
??????try?{
??????????Thread.sleep(50);
??????}?catch?(InterruptedException?e)?{
??????????e.printStackTrace();
??????}
??}
}?finally{
????unlock(lockKey,requestId);
}??
return?false;

在規(guī)定的時間，比如 500 毫秒內(nèi)，自旋不斷嘗試加鎖（說白了，就是在死循環(huán)中，不斷嘗試加鎖），如果成功則直接返回。

如果失敗，則休眠 50 毫秒，再發(fā)起新一輪的嘗試。如果到了超時時間，還未加鎖成功，則直接返回失敗。好吧，學到一招了，還有嗎？

我們都知道 redis 分布式鎖是互斥的。假如我們對某個 key 加鎖了，如果該 key 對應的鎖還沒失效，再用相同 key 去加鎖，大概率會失敗。

沒錯，大部分場景是沒問題的。為什么說是大部分場景呢？

因為還有這樣的場景：假設在某個請求中，需要獲取一顆滿足條件的菜單樹或者分類樹。我們以菜單為例，這就需要在接口中從根節(jié)點開始，遞歸遍歷出所有滿足條件的子節(jié)點，然后組裝成一顆菜單樹。

需要注意的是菜單不是一成不變的，在后臺系統(tǒng)中運營同學可以動態(tài)添加、修改和刪除菜單。為了保證在并發(fā)的情況下，每次都可能獲取最新的數(shù)據(jù)，這里可以加 redis 分布式鎖。

加 redis 分布式鎖的思路是對的。但接下來問題來了，在遞歸方法中遞歸遍歷多次，每次都是加的同一把鎖。

遞歸第一層當然是可以加鎖成功的，但遞歸第二層、第三層...第 N 層，不就會加鎖失敗了？

private?int?expireTime?=?1000;

public?void?fun(int?level,String?lockKey,String?requestId){
??try{
?????String?result?=?jedis.set(lockKey,?requestId,?"NX",?"PX",?expireTime);
?????if?("OK".equals(result))?{
????????if(level<=10){
???????????this.fun(++level,lockKey,requestId);
????????}?else?{
???????????return;
????????}
?????}
?????return;
??}?finally?{
?????unlock(lockKey,requestId);
??}
}

如果你直接這么用，看起來好像沒有問題。但最終執(zhí)行程序之后發(fā)現(xiàn)，等待你的結(jié)果只有一個：出現(xiàn)異常。

因為從根節(jié)點開始，第一層遞歸加鎖成功，還沒釋放鎖，就直接進入第二層遞歸。因為鎖名為 lockKey，并且值為 requestId 的鎖已經(jīng)存在，所以第二層遞歸大概率會加鎖失敗，然后返回到第一層。第一層接下來正常釋放鎖，然后整個遞歸方法直接返回了。

這下子，大家知道出現(xiàn)什么問題了吧？沒錯，遞歸方法其實只執(zhí)行了第一層遞歸就返回了，其他層遞歸由于加鎖失敗，根本沒法執(zhí)行。

那么這個問題該如何解決呢？

答：使用可重入鎖。

我們以 redisson 框架為例，它的內(nèi)部實現(xiàn)了可重入鎖的功能。古時候有句話說得好：為人不識陳近南，便稱英雄也枉然。

我說：分布式鎖不識 redisson，便稱好鎖也枉然。哈哈哈，只是自娛自樂一下。由此可見，redisson 在 redis 分布式鎖中的江湖地位很高。

private?int?expireTime?=?1000;

public?void?run(String?lockKey)?{
??RLock?lock?=?redisson.getLock(lockKey);
??this.fun(lock,1);
}

public?void?fun(RLock?lock,int?level){
??try{
??????lock.lock(5,?TimeUnit.SECONDS);
??????if(level<=10){
?????????this.fun(lock,++level);
??????}?else?{
?????????return;
??????}
??}?finally?{
?????lock.unlock();
??}
}

上面的代碼也許并不完美，這里只是給了一個大致的思路，如果大家有這方面需求的話，以上代碼僅供參考。接下來，聊聊 redisson 可重入鎖的實現(xiàn)原理。

if?(redis.call('exists',?KEYS[1])?==?0)?
then??
???redis.call('hset',?KEYS[1],?ARGV[2],?1);????????redis.call('pexpire',?KEYS[1],?ARGV[1]);?
???return?nil;?
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]);?
??return?nil;?
end;
return?redis.call('pttl',?KEYS[1]);

其中：

• KEYS[1]：鎖名

• ARGV[1]：過期時間

• ARGV[2]：uuid + ":" + threadId，可認為是 requestId

釋放鎖主要是通過以下腳本實現(xiàn)的：

if?(redis.call('hexists',?KEYS[1],?ARGV[3])?==?0)?
then?
??return?nil
end
local?counter?=?redis.call('hincrby',?KEYS[1],?ARGV[3],?-1);
if?(counter?>?0)?
then?
????redis.call('pexpire',?KEYS[1],?ARGV[2]);?
????return?0;?
?else?
???redis.call('del',?KEYS[1]);?
???redis.call('publish',?KEYS[2],?ARGV[1]);?
???return?1;?
end;?
return?nil

如果有大量需要寫入數(shù)據(jù)的業(yè)務場景，使用普通的 redis 分布式鎖是沒有問題的。但如果有些業(yè)務場景，寫入的操作比較少，反而有大量讀取的操作。這樣直接使用普通的 redis 分布式鎖，會不會有點浪費性能？

我們都知道，鎖的粒度越粗，多個線程搶鎖時競爭就越激烈，造成多個線程鎖等待的時間也就越長，性能也就越差。

所以，提升 redis 分布式鎖性能的第一步，就是要把鎖的粒度變細。

| 讀寫鎖

眾所周知，加鎖的目的是為了保證，在并發(fā)環(huán)境中讀寫數(shù)據(jù)的安全性，即不會出現(xiàn)數(shù)據(jù)錯誤或者不一致的情況。

但在絕大多數(shù)實際業(yè)務場景中，一般是讀數(shù)據(jù)的頻率遠遠大于寫數(shù)據(jù)。而線程間的并發(fā)讀操作是并不涉及并發(fā)安全問題，我們沒有必要給讀操作加互斥鎖，只要保證讀寫、寫寫并發(fā)操作上鎖是互斥的就行，這樣可以提升系統(tǒng)的性能。

我們以 redisson 框架為例，它內(nèi)部已經(jīng)實現(xiàn)了讀寫鎖的功能。

RReadWriteLock?readWriteLock?=?redisson.getReadWriteLock("readWriteLock");
RLock?rLock?=?readWriteLock.readLock();
try?{
????rLock.lock();
????//業(yè)務操作
}?catch?(Exception?e)?{
????log.error(e);
}?finally?{
????rLock.unlock();
}

RReadWriteLock?readWriteLock?=?redisson.getReadWriteLock("readWriteLock");
RLock?rLock?=?readWriteLock.writeLock();
try?{
????rLock.lock();
????//業(yè)務操作
}?catch?(InterruptedException?e)?{
???log.error(e);
}?finally?{
????rLock.unlock();
}

將讀鎖和寫鎖分開，最大的好處是提升讀操作的性能，因為讀和讀之間是共享的，不存在互斥性。

而我們的實際業(yè)務場景中，絕大多數(shù)數(shù)據(jù)操作都是讀操作。所以，如果提升了讀操作的性能，也就會提升整個鎖的性能。

下面總結(jié)一個讀寫鎖的特點：

• 讀與讀是共享的，不互斥

• 讀與寫互斥

• 寫與寫互斥

| 鎖分段

此外，為了減小鎖的粒度，比較常見的做法是將大鎖：分段。

在 java 中 ConcurrentHashMap，就是將數(shù)據(jù)分為 16 段，每一段都有單獨的鎖，并且處于不同鎖段的數(shù)據(jù)互不干擾，以此來提升鎖的性能。

放在實際業(yè)務場景中，我們可以這樣做：比如在秒殺扣庫存的場景中，現(xiàn)在的庫存中有 2000 個商品，用戶可以秒殺。為了防止出現(xiàn)超賣的情況，通常情況下，可以對庫存加鎖。如果有 1W 的用戶競爭同一把鎖，顯然系統(tǒng)吞吐量會非常低。

為了提升系統(tǒng)性能，我們可以將庫存分段，比如：分為 100 段，這樣每段就有 20 個商品可以參與秒殺。

如此一來，在多線程環(huán)境中，可以大大的減少鎖的沖突。以前多個線程只能同時競爭 1 把鎖，尤其在秒殺的場景中，競爭太激烈了，簡直可以用慘絕人寰來形容，其后果是導致絕大數(shù)線程在鎖等待。現(xiàn)在多個線程同時競爭 100 把鎖，等待的線程變少了，從而系統(tǒng)吞吐量也就提升了。

需要注意的地方是：將鎖分段雖說可以提升系統(tǒng)的性能，但它也會讓系統(tǒng)的復雜度提升不少。因為它需要引入額外的路由算法，跨段統(tǒng)計等功能。我們在實際業(yè)務場景中，需要綜合考慮，不是說一定要將鎖分段。

我在前面提到過，如果線程 A 加鎖成功了，但是由于業(yè)務功能耗時時間很長，超過了設置的超時時間，這時候 redis 會自動釋放線程 A 加的鎖。

有些朋友可能會說：到了超時時間，鎖被釋放了就釋放了唄，對功能又沒啥影響。

答：錯，錯，錯。對功能其實有影響。

通常我們加鎖的目的是：為了防止訪問臨界資源時，出現(xiàn)數(shù)據(jù)異常的情況。比如：線程 A 在修改數(shù)據(jù) C 的值，線程 B 也在修改數(shù)據(jù) C 的值，如果不做控制，在并發(fā)情況下，數(shù)據(jù) C 的值會出問題。

為了保證某個方法，或者段代碼的互斥性，即如果線程 A 執(zhí)行了某段代碼，是不允許其他線程在某一時刻同時執(zhí)行的，我們可以用 synchronized 關鍵字加鎖。

但這種鎖有很大的局限性，只能保證單個節(jié)點的互斥性。如果需要在多個節(jié)點中保持互斥性，就需要用 redis 分布式鎖。

此時，代碼 2 相當于裸奔的狀態(tài)，無法保證互斥性。假如它里面訪問了臨界資源，并且其他線程也訪問了該資源，可能就會出現(xiàn)數(shù)據(jù)異常的情況。（PS：我說的訪問臨界資源，不單單指讀取，還包含寫入）

那么，如何解決這個問題呢？

答：如果達到了超時時間，但業(yè)務代碼還沒執(zhí)行完，需要給鎖自動續(xù)期。

Timer?timer?=?new?Timer();?
timer.schedule(new?TimerTask()?{
????@Override
????public?void?run(Timeout?timeout)?throws?Exception?{
??????//自動續(xù)期邏輯
????}
},?10000,?TimeUnit.MILLISECONDS);

獲取鎖之后，自動開啟一個定時任務，每隔 10 秒鐘，自動刷新一次過期時間。這種機制在 redisson 框架中，有個比較霸氣的名字：watch dog，即傳說中的看門狗。

if?(redis.call('hexists',?KEYS[1],?ARGV[2])?==?1)?then?
???redis.call('pexpire',?KEYS[1],?ARGV[1]);
??return?1;?
end;
return?0;

需要注意的地方是：在實現(xiàn)自動續(xù)期功能時，還需要設置一個總的過期時間，可以跟 redisson 保持一致，設置成 30 秒。如果業(yè)務代碼到了這個總的過期時間，還沒有執(zhí)行完，就不再自動續(xù)期了。

自動續(xù)期的功能是獲取鎖之后開啟一個定時任務，每隔 10 秒判斷一下鎖是否存在，如果存在，則刷新過期時間。如果續(xù)期 3 次，也就是 30 秒之后，業(yè)務方法還是沒有執(zhí)行完，就不再續(xù)期了。

上面花了這么多篇幅介紹的內(nèi)容，對單個 redis 實例是沒有問題的。

but，如果 redis 存在多個實例。比如：做了主從，或者使用了哨兵模式，基于 redis 的分布式鎖的功能，就會出現(xiàn)問題。

本來是和諧共處，相安無事的。redis 加鎖操作，都在 master 上進行，加鎖成功后，再異步同步給所有的 slave。

如果有個鎖 A 比較悲催，剛加鎖成功 master 就掛了，還沒來得及同步到 slave1。

這樣會導致新 master 節(jié)點中的鎖 A 丟失了。后面，如果有新的線程，使用鎖 A 加鎖，依然可以成功，分布式鎖失效了。

那么，如何解決這個問題呢？

答：redisson 框架為了解決這個問題，提供了一個專門的類：RedissonRedLock，使用了 Redlock 算法。

RedissonRedLock 解決問題的思路如下：

• 需要搭建幾套相互獨立的 redis 環(huán)境，假如我們在這里搭建了 5 套。

• 每套環(huán)境都有一個 redisson node 節(jié)點。

• 多個 redisson node 節(jié)點組成了 RedissonRedLock。

• 環(huán)境包含：單機、主從、哨兵和集群模式，可以是一種或者多種混合。

RedissonRedLock 加鎖過程如下：

• 獲取所有的 redisson node 節(jié)點信息，循環(huán)向所有的 redisson node 節(jié)點加鎖，假設節(jié)點數(shù)為 N，例子中 N 等于 5。

• 如果在 N 個節(jié)點當中，有 N/2+1 個節(jié)點加鎖成功了，那么整個 RedissonRedLock 加鎖是成功的。

• 如果在 N 個節(jié)點當中，小于 N/2+1 個節(jié)點加鎖成功，那么整個 RedissonRedLock 加鎖是失敗的。

• 如果中途發(fā)現(xiàn)各個節(jié)點加鎖的總耗時，大于等于設置的最大等待時間，則直接返回失敗。

從上面可以看出，使用 Redlock 算法，確實能解決多實例場景中，假如 master 節(jié)點掛了，導致分布式鎖失效的問題。

但也引出了一些新問題，比如：

• 需要額外搭建多套環(huán)境，申請更多的資源，需要評估一下成本和性價比。

• 如果有 N 個 redisson node 節(jié)點，需要加鎖 N 次，最少也需要加鎖 N/2+1 次，才知道 redlock 加鎖是否成功。顯然，增加了額外的時間成本，有點得不償失。

由此可見，在實際業(yè)務場景，尤其是高并發(fā)業(yè)務中，RedissonRedLock 其實使用的并不多。在分布式環(huán)境中，CAP 是繞不過去的。

CAP 指的是在一個分布式系統(tǒng)中：

• 一致性（Consistency）

• 可用性（Availability）

• 分區(qū)容錯性（Partition tolerance）

這三個要素最多只能同時實現(xiàn)兩點，不可能三者兼顧。

如果你的實際業(yè)務場景，更需要的是保證數(shù)據(jù)一致性。那么請使用 CP 類型的分布式鎖，比如：zookeeper，它是基于磁盤的，性能可能沒那么好，但數(shù)據(jù)一般不會丟。

如果你的實際業(yè)務場景，更需要的是保證數(shù)據(jù)高可用性。那么請使用 AP 類型的分布式鎖，比如：redis，它是基于內(nèi)存的，性能比較好，但有丟失數(shù)據(jù)的風險。

其實，在我們絕大多數(shù)分布式業(yè)務場景中，使用 redis 分布式鎖就夠了，真的別太較真。因為數(shù)據(jù)不一致問題，可以通過最終一致性方案解決。但如果系統(tǒng)不可用了，對用戶來說是暴擊一萬點傷害。