為什么用分布式鎖？在討論這個問題之前，我們先來看一個業務場景。

[[270723]]

圖片來自 Pexels

為什么用分布式鎖？

系統 A 是一個電商系統，目前是一臺機器部署，系統中有一個用戶下訂單的接口，但是用戶下訂單之前一定要去檢查一下庫存，確保庫存足夠了才會給用戶下單。

由于系統有一定的并發，所以會預先將商品的庫存保存在 Redis 中，用戶下單的時候會更新 Redis 的庫存。

此時系統架構如下：

但是這樣一來會產生一個問題：假如某個時刻，Redis 里面的某個商品庫存為 1。

此時兩個請求同時到來，其中一個請求執行到上圖的第 3 步，更新數據庫的庫存為 0，但是第 4 步還沒有執行。

而另外一個請求執行到了第 2 步，發現庫存還是 1，就繼續執行第 3 步。這樣的結果，是導致賣出了 2 個商品，然而其實庫存只有 1 個。

很明顯不對啊！這就是典型的庫存超賣問題。此時，我們很容易想到解決方案：用鎖把 2、3、4 步鎖住，讓他們執行完之后，另一個線程才能進來執行第 2 步。

按照上面的圖，在執行第 2 步時，使用 Java 提供的 Synchronized 或者 ReentrantLock 來鎖住，然后在第 4 步執行完之后才釋放鎖。

這樣一來，2、3、4 這 3 個步驟就被“鎖”住了，多個線程之間只能串行化執行。

但是好景不長，整個系統的并發飆升，一臺機器扛不住了。現在要增加一臺機器，如下圖：

增加機器之后，系統變成上圖所示，我的天！假設此時兩個用戶的請求同時到來，但是落在了不同的機器上，那么這兩個請求是可以同時執行了，還是會出現庫存超賣的問題。

為什么呢？因為上圖中的兩個 A 系統，運行在兩個不同的 JVM 里面，他們加的鎖只對屬于自己 JVM 里面的線程有效，對于其他 JVM 的線程是無效的。

因此，這里的問題是：Java 提供的原生鎖機制在多機部署場景下失效了，這是因為兩臺機器加的鎖不是同一個鎖（兩個鎖在不同的 JVM 里面）。

那么，我們只要保證兩臺機器加的鎖是同一個鎖，問題不就解決了嗎？此時，就該分布式鎖隆重登場了。

分布式鎖的思路是：在整個系統提供一個全局、唯一的獲取鎖的“東西”，然后每個系統在需要加鎖時，都去問這個“東西”拿到一把鎖，這樣不同的系統拿到的就可以認為是同一把鎖。

至于這個“東西”，可以是 Redis、Zookeeper，也可以是數據庫。文字描述不太直觀，我們來看下圖：

通過上面的分析，我們知道了庫存超賣場景在分布式部署系統的情況下使用 Java 原生的鎖機制無法保證線程安全，所以我們需要用到分布式鎖的方案。

那么，如何實現分布式鎖呢？接著往下看！

基于 Redis 實現分布式鎖

上面分析為啥要使用分布式鎖了，這里我們來具體看看分布式鎖落地的時候應該怎么樣處理。

①常見的一種方案就是使用 Redis 做分布式鎖

使用 Redis 做分布式鎖的思路大概是這樣的：在 Redis 中設置一個值表示加了鎖，然后釋放鎖的時候就把這個 Key 刪除。

具體代碼是這樣的：

// 獲取鎖 
// NX是指如果key不存在就成功，key存在返回false，PX可以指定過期時間 
SET anyLock unique_value NX PX 30000 
 
 
// 釋放鎖：通過執行一段lua腳本 
// 釋放鎖涉及到兩條指令，這兩條指令不是原子性的 
// 需要用到redis的lua腳本支持特性，redis執行lua腳本是原子性的 
if redis.call("get",KEYS[1]) == ARGV[1] then 
return redis.call("del",KEYS[1]) 
else 
return 0 
end 

這種方式有幾大要點：

• 一定要用 SET key value NX PX milliseconds 命令。如果不用，先設置了值，再設置過期時間，這個不是原子性操作，有可能在設置過期時間之前宕機，會造成死鎖(Key ***存在)
• Value 要具有唯一性。這個是為了在解鎖的時候，需要驗證 Value 是和加鎖的一致才刪除 Key。

這時避免了一種情況：假設 A 獲取了鎖，過期時間 30s，此時 35s 之后，鎖已經自動釋放了，A 去釋放鎖，但是此時可能 B 獲取了鎖。A 客戶端就不能刪除 B 的鎖了。

除了要考慮客戶端要怎么實現分布式鎖之外，還需要考慮 Redis 的部署問題。

Redis 有 3 種部署方式：

• 單機模式
• Master-Slave+Sentinel 選舉模式
• Redis Cluster 模式

使用 Redis 做分布式鎖的缺點在于：如果采用單機部署模式，會存在單點問題，只要 Redis 故障了。加鎖就不行了。

采用 Master-Slave 模式，加鎖的時候只對一個節點加鎖，即便通過 Sentinel 做了高可用，但是如果 Master 節點故障了，發生主從切換，此時就會有可能出現鎖丟失的問題。

基于以上的考慮，Redis 的作者也考慮到這個問題，他提出了一個 RedLock 的算法。

這個算法的意思大概是這樣的：假設 Redis 的部署模式是 Redis Cluster，總共有 5 個 Master 節點。

通過以下步驟獲取一把鎖：

• 獲取當前時間戳，單位是毫秒。
• 輪流嘗試在每個 Master 節點上創建鎖，過期時間設置較短，一般就幾十毫秒。
• 嘗試在大多數節點上建立一個鎖，比如 5 個節點就要求是 3 個節點（n / 2 +1）。
• 客戶端計算建立好鎖的時間，如果建立鎖的時間小于超時時間，就算建立成功了。
• 要是鎖建立失敗了，那么就依次刪除這個鎖。
• 只要別人建立了一把分布式鎖，你就得不斷輪詢去嘗試獲取鎖。

但是這樣的這種算法還是頗具爭議的，可能還會存在不少的問題，無法保證加鎖的過程一定正確。

②另一種方式：Redisson

此外，實現 Redis 的分布式鎖，除了自己基于 Redis Client 原生 API 來實現之外，還可以使用開源框架：Redission。

Redisson 是一個企業級的開源 Redis Client，也提供了分布式鎖的支持。我也非常推薦大家使用，為什么呢？

回想一下上面說的，如果自己寫代碼來通過 Redis 設置一個值，是通過下面這個命令設置的：

SET anyLock unique_value NX PX 30000 

這里設置的超時時間是 30s，假如我超過 30s 都還沒有完成業務邏輯的情況下，Key 會過期，其他線程有可能會獲取到鎖。

這樣一來的話，***個線程還沒執行完業務邏輯，第二個線程進來了也會出現線程安全問題。

所以我們還需要額外的去維護這個過期時間，太麻煩了~我們來看看 Redisson 是怎么實現的？

先感受一下使用 Redission 的爽：

Config config = new Config(); 
config.useClusterServers() 
.addNodeAddress("redis://192.168.31.101:7001") 
.addNodeAddress("redis://192.168.31.101:7002") 
.addNodeAddress("redis://192.168.31.101:7003") 
.addNodeAddress("redis://192.168.31.102:7001") 
.addNodeAddress("redis://192.168.31.102:7002") 
.addNodeAddress("redis://192.168.31.102:7003"); 
 
RedissonClient redisson = Redisson.create(config); 
 
 
RLock lock = redisson.getLock("anyLock"); 
lock.lock(); 
lock.unlock(); 

就是這么簡單，我們只需要通過它的 API 中的 Lock 和 Unlock 即可完成分布式鎖，他幫我們考慮了很多細節：

• Redisson 所有指令都通過 Lua 腳本執行，Redis 支持 Lua 腳本原子性執行。
• Redisson 設置一個 Key 的默認過期時間為 30s，如果某個客戶端持有一個鎖超過了 30s 怎么辦？
• Redisson 中有一個 Watchdog 的概念，翻譯過來就是看門狗，它會在你獲取鎖之后，每隔 10s 幫你把 Key 的超時時間設為 30s。

這樣的話，就算一直持有鎖也不會出現 Key 過期了，其他線程獲取到鎖的問題了。

• Redisson 的“看門狗”邏輯保證了沒有死鎖發生。(如果機器宕機了，看門狗也就沒了。此時就不會延長 Key 的過期時間，到了 30s 之后就會自動過期了，其他線程可以獲取到鎖)

這里稍微貼出來其實現代碼：

// 加鎖邏輯 
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) { 
    if (leaseTime != -1) { 
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); 
    } 
    // 調用一段lua腳本，設置一些key、過期時間 
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); 
    ttlRemainingFuture.addListener(new FutureListener<Long>() { 
        @Override 
        public void operationComplete(Future<Long> future) throws Exception { 
            if (!future.isSuccess()) { 
                return; 
            } 
 
            Long ttlRemaining = future.getNow(); 
            // lock acquired 
            if (ttlRemaining == null) { 
                // 看門狗邏輯 
                scheduleExpirationRenewal(threadId); 
            } 
        } 
    }); 
    return ttlRemainingFuture; 
} 
 
 
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { 
    internalLockLeaseTime = unit.toMillis(leaseTime); 
 
    return commandExecutor.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]); " + 
                  "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]);", 
                Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); 
} 
 
 
 
// 看門狗最終會調用了這里 
private void scheduleExpirationRenewal(final long threadId) { 
    if (expirationRenewalMap.containsKey(getEntryName())) { 
        return; 
    } 
 
    // 這個任務會延遲10s執行 
    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { 
        @Override 
        public void run(Timeout timeout) throws Exception { 
 
            // 這個操作會將key的過期時間重新設置為30s 
            RFuture<Boolean> future = renewExpirationAsync(threadId); 
 
            future.addListener(new FutureListener<Boolean>() { 
                @Override 
                public void operationComplete(Future<Boolean> future) throws Exception { 
                    expirationRenewalMap.remove(getEntryName()); 
                    if (!future.isSuccess()) { 
                        log.error("Can't update lock " + getName() + " expiration", future.cause()); 
                        return; 
                    } 
 
                    if (future.getNow()) { 
                        // reschedule itself 
                        // 通過遞歸調用本方法，***循環延長過期時間 
                        scheduleExpirationRenewal(threadId); 
                    } 
                } 
            }); 
        } 
 
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); 
 
    if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) { 
        task.cancel(); 
    } 
} 

另外，Redisson 還提供了對 Redlock 算法的支持，它的用法也很簡單：

RedissonClient redisson = Redisson.create(config); 
RLock lock1 = redisson.getFairLock("lock1"); 
RLock lock2 = redisson.getFairLock("lock2"); 
RLock lock3 = redisson.getFairLock("lock3"); 
RedissonRedLock multiLock = new RedissonRedLock(lock1, lock2, lock3); 
multiLock.lock(); 
multiLock.unlock(); 

小結：本節分析了使用 Redis 作為分布式鎖的具體落地方案以及其一些局限性，然后介紹了一個 Redis 的客戶端框架 Redisson，這也是我推薦大家使用的，比自己寫代碼實現會少 Care 很多細節。

基于 Zookeeper 實現分布式鎖

常見的分布式鎖實現方案里面，除了使用 Redis 來實現之外，使用 Zookeeper 也可以實現分布式鎖。

在介紹 Zookeeper(下文用 ZK 代替)實現分布式鎖的機制之前，先粗略介紹一下 ZK 是什么東西：ZK 是一種提供配置管理、分布式協同以及命名的中心化服務。

ZK 的模型是這樣的：ZK 包含一系列的節點，叫做 Znode，就好像文件系統一樣，每個 Znode 表示一個目錄。

然后 Znode 有一些特性：

• 有序節點：假如當前有一個父節點為 /lock，我們可以在這個父節點下面創建子節點，ZK 提供了一個可選的有序特性。

例如我們可以創建子節點“/lock/node-”并且指明有序，那么 ZK 在生成子節點時會根據當前的子節點數量自動添加整數序號。

也就是說，如果是***個創建的子節點，那么生成的子節點為 /lock/node-0000000000，下一個節點則為 /lock/node-0000000001，依次類推。

• 臨時節點：客戶端可以建立一個臨時節點，在會話結束或者會話超時后，ZK 會自動刪除該節點。
• 事件監聽：在讀取數據時，我們可以同時對節點設置事件監聽，當節點數據或結構變化時，ZK 會通知客戶端。

當前 ZK 有如下四種事件：

• 節點創建
• 節點刪除
• 節點數據修改
• 子節點變更

基于以上的一些 ZK 的特性，我們很容易得出使用 ZK 實現分布式鎖的落地方案：

• 使用 ZK 的臨時節點和有序節點，每個線程獲取鎖就是在 ZK 創建一個臨時有序的節點，比如在 /lock/ 目錄下。
• 創建節點成功后，獲取 /lock 目錄下的所有臨時節點，再判斷當前線程創建的節點是否是所有的節點的序號最小的節點。
• 如果當前線程創建的節點是所有節點序號最小的節點，則認為獲取鎖成功。
• 如果當前線程創建的節點不是所有節點序號最小的節點，則對節點序號的前一個節點添加一個事件監聽。

比如當前線程獲取到的節點序號為 /lock/003，然后所有的節點列表為[/lock/001，/lock/002，/lock/003]，則對 /lock/002 這個節點添加一個事件監聽器。

如果鎖釋放了，會喚醒下一個序號的節點，然后重新執行第 3 步，判斷是否自己的節點序號是最小。

比如 /lock/001 釋放了，/lock/002 監聽到時間，此時節點集合為[/lock/002，/lock/003]，則 /lock/002 為最小序號節點，獲取到鎖。

整個過程如下：

具體的實現思路就是這樣，至于代碼怎么寫，這里比較復雜就不貼出來了。

Curator 介紹

Curator 是一個 ZK 的開源客戶端，也提供了分布式鎖的實現。它的使用方式也比較簡單：

InterProcessMutex interProcessMutex = new InterProcessMutex(client,"/anyLock"); 
interProcessMutex.acquire(); 
interProcessMutex.release(); 

其實現分布式鎖的核心源碼如下：

private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception 
{ 
    boolean  haveTheLock = false; 
    boolean  doDelete = false; 
    try { 
        if ( revocable.get() != null ) { 
            client.getData().usingWatcher(revocableWatcher).forPath(ourPath); 
        } 
 
        while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock ) { 
            // 獲取當前所有節點排序后的集合 
            List<String>        children = getSortedChildren(); 
            // 獲取當前節點的名稱 
            String              sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash 
            // 判斷當前節點是否是最小的節點 
            PredicateResults    predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases); 
            if ( predicateResults.getsTheLock() ) { 
                // 獲取到鎖 
                haveTheLock = true; 
            } else { 
                // 沒獲取到鎖，對當前節點的上一個節點注冊一個監聽器 
                String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch(); 
                synchronized(this){ 
                    Stat stat = client.checkExists().usingWatcher(watcher).forPath(previousSequencePath); 
                    if ( stat != null ){ 
                        if ( millisToWait != null ){ 
                            millisToWait -= (System.currentTimeMillis() - startMillis); 
                            startMillis = System.currentTimeMillis(); 
                            if ( millisToWait <= 0 ){ 
                                doDelete = true;    // timed out - delete our node 
                                break; 
                            } 
                            wait(millisToWait); 
                        }else{ 
                            wait(); 
                        } 
                    } 
                } 
                // else it may have been deleted (i.e. lock released). Try to acquire again 
            } 
        } 
    } 
    catch ( Exception e ) { 
        doDelete = true; 
        throw e; 
    } finally{ 
        if ( doDelete ){ 
            deleteOurPath(ourPath); 
        } 
    } 
    return haveTheLock; 
} 

其實 Curator 實現分布式鎖的底層原理和上面分析的是差不多的。這里我們用一張圖詳細描述其原理：

小結：本節介紹了 ZK 實現分布式鎖的方案以及 ZK 的開源客戶端的基本使用，簡要的介紹了其實現原理。

兩種方案的優缺點比較

學完了兩種分布式鎖的實現方案之后，本節需要討論的是 Redis 和 ZK 的實現方案中各自的優缺點。

對于 Redis 的分布式鎖而言，它有以下缺點：

• 它獲取鎖的方式簡單粗暴，獲取不到鎖直接不斷嘗試獲取鎖，比較消耗性能。
• 另外來說的話，Redis 的設計定位決定了它的數據并不是強一致性的，在某些極端情況下，可能會出現問題。鎖的模型不夠健壯。
• 即便使用 Redlock 算法來實現，在某些復雜場景下，也無法保證其實現 100% 沒有問題，關于 Redlock 的討論可以看 How to do distributed locking。
• Redis 分布式鎖，其實需要自己不斷去嘗試獲取鎖，比較消耗性能。

但是另一方面使用 Redis 實現分布式鎖在很多企業中非常常見，而且大部分情況下都不會遇到所謂的“極端復雜場景”。

所以使用 Redis 作為分布式鎖也不失為一種好的方案，最重要的一點是 Redis 的性能很高，可以支撐高并發的獲取、釋放鎖操作。

對于 ZK 分布式鎖而言:

• ZK 天生設計定位就是分布式協調，強一致性。鎖的模型健壯、簡單易用、適合做分布式鎖。
• 如果獲取不到鎖，只需要添加一個監聽器就可以了，不用一直輪詢，性能消耗較小。

但是 ZK 也有其缺點：如果有較多的客戶端頻繁的申請加鎖、釋放鎖，對于 ZK 集群的壓力會比較大。

小結：綜上所述，Redis 和 ZK 都有其優缺點。我們在做技術選型的時候可以根據這些問題作為參考因素。

一些建議

通過前面的分析，實現分布式鎖的兩種常見方案：Redis 和 ZK，他們各有千秋。應該如何選型呢？

就個人而言的話，我比較推崇 ZK 實現的鎖：因為 Redis 是有可能存在隱患的，可能會導致數據不對的情況。但是，怎么選用要看具體在公司的場景了。

如果公司里面有 ZK 集群條件，優先選用 ZK 實現，但是如果說公司里面只有 Redis 集群，沒有條件搭建 ZK 集群。

那么其實用 Redis 來實現也可以，另外還可能是系統設計者考慮到了系統已經有 Redis，但是又不希望再次引入一些外部依賴的情況下，可以選用 Redis。這個是要系統設計者基于架構來考慮了。

中華石杉：十余年 BAT 架構經驗，一線互聯網公司技術總監。帶領上百人團隊開發過多個億級流量高并發系統。現將多年工作中積累下的研究手稿、經驗總結整理成文，傾囊相授。微信公眾號：石杉的架構筆記（ID：shishan100）。