[[354141]]

本文轉載自微信公眾號「故里學Java」，作者故里 。轉載本文請聯系故里學Java公眾號。  

記錄一下今天的文章開始寫的時間00:53，夜深人靜了，我們來學一下分布式鎖，我們要悄悄地學習，然后經驗所有人。

什么是分布式鎖?分布式鎖又可以解決哪些問題呢?

在我們的系統還沒有使用分布式架構的時候，我們可以用同步鎖或者Lock鎖，來保證多線程并發的時候，同一時間只有一個線程修改共享變量或者執行代碼塊，但是當我們現在大部分系統都是分布式集群部署的，單純的同步鎖和Lock鎖只能保證單個實例上的數據一致性，多實例就失去了作用。

這個時候就需要使用分布式鎖來保證共享資源的原子性，比如我們電商系統里面的扣減庫存，當單量小的時候問題不大，如果單量很大，同一時間多個實例都在并發處理扣減庫存的業務的時候，就可能存在超賣的問題。

分布式鎖的實現?

常見的分布式鎖有數據庫實現分布式鎖、Zookeeper實現分布式鎖、Redis實現分布式鎖、Redisson實現。其中數據庫實現分布式鎖比較簡單，也很容易理解，直接基于數據庫實現就可以了，在一些分布式的業務中也經常使用，但是這種方式也是效率最低的，一般是不使用的，我們就著重介紹一下其他三種方式的實現。

Zookeeper實現分布式鎖

使用Zookeeper來實現分布式鎖就比較常見，比如很多項目就使用Zookeeper作為分布式注冊中心，就喜歡用Zookeeper來實現分布式鎖，這主要是借助于Zookeeper的兩大特性：順序臨時節點、Watch機制。

順序臨時節點：熟悉Zookeeper的同學都知道，Zookeeper提供了多層級的節點命名空間，每個節點都是用斜杠分隔的路徑來表示，類似于我們的文件夾。節點又分為持久節點和臨時節點，節點還可以標記為有序，當節點被標記為有序性，這個節點就具有順序自增的特點，我們就可以借助這個特點來創建我們所需的節點。

Watch機制：Watch機制是Zookeeper另一個重要的特性，我們可以在指定節點上注冊一些Watcher，在一些特定的事情觸發的時候，通知用戶這個事件。

Zookeeper實現分布式鎖的過程

我們先創建一個持久節點作為父節點，每當需要訪問創建分布式鎖的時候，就在這個父節點下創建相應的臨時的順序子節點，以臨時節點名稱、父節點名稱和順序號組成特點的名稱。在建立子節點后，對父節點下以這個這個子節點名稱開頭的子節點進行排序，判斷剛建立的節點順序號是不是最小的，如果是最小的則獲取鎖，如果不是最小節點，則阻塞等待鎖，并且在獲取該節點的上一順序節點注冊Watcher，等待節點對應的操作獲得鎖。

當業務處理完之后，刪除該節點，關閉zk，進而觸發Watcher，釋放該鎖。

上圖就是就是嚴格按照順序訪問的分布式鎖實現，更多的時候我們引入一些框架來幫助我們實現，比如最常用的Curator框架，代碼如下：

InterProcessMutex lock = new InterProcessMutex(client, lockPath); 
if ( lock.acquire(maxWait, waitUnit) ) { 
    try { 
        // 業務處理 
    } 
    finally{ 
        lock.release(); 
    } 
} 

Zookeeper來實現分布式鎖天然的優勢就是，Zookeeper是集群實現的，我們生產環境一般也是集群部署的，可以避免單點問題，穩定性較好，能保證每次操作都可以釋放鎖。

缺點就是，頻繁的創建刪除節點，加上注冊watch事件，對于zookeeper集群的壓力比較大，性能這一塊也比不上Redis實現的分布式鎖。

Redis實現分布式鎖

Redis實現的分布式鎖，最為復雜，但是性能確是最佳的，所以在對性能要求更高的系統里，我們都選擇使用Redis來實現分布式鎖。利用Redis實現分布式鎖，一般都是使用SETNX實現，舉個簡單的例子：

public static boolean getDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) { 
 
    String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime); 
 
    if ("OK".equals(result)) { 
         return true; 
    } 
    return false; 
} 

SETNX方法保證設置鎖和鎖過期時間的原子性，但是對于鎖的過期時間設置我們要注意，如果執行業務

時間比較長，我們設置的過期時間又比較短的情況下就會造成，業務還沒執行完，鎖已釋放的問題。所以我們需要根據實際業務處理來評估設置鎖的過期時間，來保證業務可以正常的處理完。

Redisson實現分布式鎖

Redisson是架設在Redis基礎上的一個Java駐內存數據網格。Redisson在基于NIO的Netty框架上，充分的利用了Redis鍵值數據庫提供的一系列優勢，在Java實用工具包中常用接口的基礎上，為使用者提供了一系列具有分布式特性的常用工具類。性能也比我們常用的jedis好一些。

Redisson不管是單節點模式還是集群模式，都很好的實現了分布式鎖，一般用的多的都是集群模式，在集群模式下，Redisson使用RedLock算法，很好的處理了Master節點宕機時切換到另外一個Master節點過程中多個應用獲得鎖。

Redisson集群模式獲取鎖的實現就是，在不同節點上獲取鎖，每個節點上獲取鎖都有超時時間，如果獲取鎖超時就認為這個節點不可用，當成功獲取鎖的個數超過Redis節點的半數，且獲取鎖消耗的時間還沒超過鎖過期時間，則認為獲取鎖成功。獲取鎖成功后重新計算鎖釋放時間，由原來的鎖釋放時間減去獲取鎖消耗的時間，如果最終獲取鎖失敗，已經獲取鎖成功的節點也會釋放鎖。

具體的代碼實現：

引入依賴

<dependency> 
    <groupId>org.redisson</groupId> 
    <artifactId>redisson</artifactId> 
    <version>3.13.1</version> 
</dependency> 

Redisson配置文件：

@Bean 
public RedissonClient redissonClient() { 
    Config config = new Config(); 
    config.useClusterServers() 
            .setScanInterval(3000) // 集群狀態掃描間隔時間，單位是毫秒 
            .addNodeAddress("redis://192.168.0.1:6379).setPassword("666") 
            .addNodeAddress("redis://192.168.0.2:6379").setPassword("666") 
            .addNodeAddress("redis://192.168.0.3:6379") 
            .setPassword("666"); 
    return Redisson.create(config); 
} 

獲取鎖操作：

long waitTimeout = 10; 
long leaseTime = 1; 
RLock lock1 = redissonClient1.getLock("lock1"); 
RLock lock2 = redissonClient2.getLock("lock2"); 
RLock lock3 = redissonClient3.getLock("lock3"); 
 
RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3); 
 
redLock.trylock(waitTimeout,leaseTime,TimeUnit.SECONDS); 
try{ 
    //... 
}finally{ 
    redLock.unlock(); 
} 

總結

實現分布式鎖的方式不止這三種，最簡單的就是數據庫實現，Zookeeper實現也相對比較簡單，但是性能最好的還是Redis實現，但是可靠性方面，Zookeeper基于分布式集群，具有天然的優勢，可靠性相對更高。如果業務場景對性能要求不是很高的時候，優先使用Zookeeper實現分布式鎖。