如何使用分布式鎖

本篇內容主要講解“如何使用分布式鎖”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“如何使用分布式鎖”吧!

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

在我們的系統還沒有使用分布式架構的時候，我們可以用同步鎖或者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實現分布式鎖。

到此，相信大家對“如何使用分布式鎖”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！