點擊上方“朱小廝的博客”，選擇“設為星標”

后臺回復"書"，獲取

后臺回復“k8s”，可領取k8s資料

來源：r6d.cn/W3Gy

“

提到鎖大家肯定有了解，像 Synchronized、ReentrantLock，在單進程情況下，多個線程訪問同一資源，可以用它們來保證線程的安全性。

不過目前互聯網項目越來越多的項目采用集群部署，也就是分布式情況，這兩種鎖就有些不夠用了。

來兩張圖舉例說明下，本地鎖的情況下：

分布式鎖情況下：

就其思想來說，就是一種“我全都要”的思想，所有服務都到一個統一的地方來取鎖，只有取到鎖的才能繼續執行下去。

說完思想，下面來說一下具體的實現。

Redis 實現

為實現分布式鎖，在 Redis 中存在 SETNX key value 命令，意為 set if not exists（如果不存在該 key，才去 set 值），就比如說是張三去上廁所，看廁所門鎖著，他就不進去了，廁所門開著他才去。

可以看到，第一次 set 返回了 1，表示成功，但是第二次返回 0，表示 set 失敗，因為已經存在這個 key 了。

當然只靠 setnx 這個命令可以嗎？當然是不行的，試想一種情況，張三在廁所里，但他在里面一直沒有釋放，一直在里面蹲著，那外面人想去廁所全部都去不了，都想錘死他了。

Redis 同理，假設已經進行了加鎖，但是因為宕機或者出現異常未釋放鎖，就造成了所謂的“死鎖”。

聰明的你們肯定早都想到了，為它設置過期時間不就好了，可以 SETEX key seconds value 命令，為指定 key 設置過期時間，單位為秒。

但這樣又有另一個問題，我剛加鎖成功，還沒設置過期時間，Redis 宕機了不就又死鎖了，所以說要保證原子性吖，要么一起成功，要么一起失敗。

當然我們能想到的 Redis 肯定早都為你實現好了，在 Redis 2.8 的版本后，Redis 就為我們提供了一條組合命令 SET key value ex seconds nx，加鎖的同時設置過期時間。

就好比是公司規定每人最多只能在廁所呆 2 分鐘，不管釋放沒釋放完都得出來，這樣就解決了“死鎖”問題。

但這樣就沒有問題了嗎？怎么可能。

試想又一種情況，廁所門肯定只能從里面開啊，張三上完廁所后張四進去鎖上門，但是外面人以為還是張三在里面，而且已經過了 3 分鐘了，就直接把門給撬開了，一看里面卻是張四，這就很尷尬啊。

換成 Redis 就是說比如一個業務執行時間很長，鎖已經自己過期了，別人已經設置了新的鎖，但是當業務執行完之后直接釋放鎖，就有可能是刪除了別人加的鎖，這不是亂套了嗎。

所以在加鎖時候，要設一個隨機值，在刪除鎖時進行比對，如果是自己的鎖，才刪除。

多說無益，煩人，直接上代碼：

//基于jedis和lua腳本來實現 privatestaticfinal?String?LOCK_SUCCESS?=?"OK"; privatestaticfinal?Long?RELEASE_SUCCESS?=?1L; privatestaticfinal?String?SET_IF_NOT_EXIST?=?"NX"; privatestaticfinal?String?SET_WITH_EXPIRE_TIME?=?"PX";@Override public?String?acquire()?{try?{//?獲取鎖的超時時間，超過這個時間則放棄獲取鎖long?end?=?System.currentTimeMillis()?+?acquireTimeout;//?隨機生成一個?valueString?requireToken?=?UUID.randomUUID().toString();while?(System.currentTimeMillis()?<?end)?{String?result?=?jedis.set(lockKey,?requireToken,?SET_IF_NOT_EXIST,?SET_WITH_EXPIRE_TIME,?expireTime);if?(LOCK_SUCCESS.equals(result))?{return?requireToken;}try?{Thread.sleep(100);}?catch?(InterruptedException?e)?{Thread.currentThread().interrupt();}}}?catch?(Exception?e)?{log.error("acquire?lock?due?to?error",?e);}returnnull; }@Override public?boolean?release(String?identify)?{if?(identify?==?null)?{returnfalse;}//通過lua腳本進行比對刪除操作，保證原子性String?script?=?"if?redis.call('get',?KEYS[1])?==?ARGV[1]?then?return?redis.call('del',?KEYS[1])?else?return?0?end";Object?result?=?new?Object();try?{result?=?jedis.eval(script,?Collections.singletonList(lockKey),Collections.singletonList(identify));if?(RELEASE_SUCCESS.equals(result))?{log.info("release?lock?success,?requestToken:{}",?identify);returntrue;}}?catch?(Exception?e)?{log.error("release?lock?due?to?error",?e);}?finally?{if?(jedis?!=?null)?{jedis.close();}}log.info("release?lock?failed,?requestToken:{},?result:{}",?identify,?result);returnfalse; }

思考：加鎖和釋放鎖的原子性可以用 lua 腳本來保證，那鎖的自動續期改如何實現呢？

Redisson 實現

Redisson 顧名思義，Redis 的兒子，本質上還是 Redis 加鎖，不過是對 Redis 做了很多封裝，它不僅提供了一系列的分布式的 Java 常用對象，還提供了許多分布式服務。

在引入 Redisson 的依賴后，就可以直接進行調用：

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

先來一段 Redisson 的加鎖代碼：

private?void?test()?{//分布式鎖名??鎖的粒度越細，性能越好RLock?lock?=?redissonClient.getLock("test_lock");lock.lock();try?{//具體業務......}?finally?{lock.unlock();} }

就是這么簡單，使用方法 jdk 的 ReentrantLock 差不多，并且也支持 ReadWriteLock（讀寫鎖）、Reentrant Lock（可重入鎖）、Fair Lock（公平鎖）、RedLock（紅鎖）等各種鎖，詳細可以參照redisson官方文檔來查看。

那么 Redisson 到底有哪些優勢呢？鎖的自動續期（默認都是 30 秒），如果業務超長，運行期間會自動給鎖續上新的 30s，不用擔心業務執行時間超長而鎖被自動刪掉。

加鎖的業務只要運行完成，就不會給當前續期，即便不手動解鎖，鎖默認在 30s 后刪除，不會造成死鎖問題。

前面也提到了鎖的自動續期，我們來看看 Redisson 是如何來實現的。

先說明一下，這里主要講的是 Redisson 中的 RLock，也就是可重入鎖，有兩種實現方法：

//?最常見的使用方法 lock.lock();//?加鎖以后10秒鐘自動解鎖 //?無需調用unlock方法手動解鎖 lock.lock(10,?TimeUnit.SECONDS);

而只有無參的方法是提供鎖的自動續期操作的，內部使用的是“看門狗”機制，我們來看一看源碼。

不管是空參還是帶參方法，它們都調用的是同一個 lock 方法，未傳參的話時間傳了一個 -1，而帶參的方法傳過去的就是實際傳入的時間。

繼續點進 scheduleExpirationRenewal 方法：

點進 renewExpiration 方法：

總結一下，就是當我們指定鎖過期時間，那么鎖到時間就會自動釋放。如果沒有指定鎖過期時間，就使用看門狗的默認時間 30s，只要占鎖成功，就會啟動一個定時任務，每隔 10s 給鎖設置新的過期時間，時間為看門狗的默認時間，直到鎖釋放。

小結：雖然 lock() 有自動續鎖機制，但是開發中還是推薦使用 lock(time，timeUnit)，因為它省掉了整個續期帶來的性能損，可以設置過期時間長一點，搭配 unlock()。

若業務執行完成，會手動釋放鎖，若是業務執行超時，那一般我們服務也都會設置業務超時時間，就直接報錯了，報錯后就會通過設置的過期時間來釋放鎖。

public?void?test()?{RLock?lock?=?redissonClient.getLock("test_lock");lock.lock(30,?TimeUnit.SECONDS);try?{//.......具體業務}?finally?{//手動釋放鎖lock.unlock();} }

基于 Zookeeper 來實現分布式鎖

很多小伙伴都知道在分布式系統中，可以用 ZK 來做注冊中心，但其實在除了做祖冊中心以外，用 ZK 來做分布式鎖也是很常見的一種方案。

先來看一下 ZK 中是如何創建一個節點的？ZK 中存在 create [-s] [-e] ?path [data] 命令，-s 為創建有序節點，-e 創建臨時節點。

這樣就創建了一個父節點并為父節點創建了一個子節點，組合命令意為創建一個臨時的有序節點。

而 ZK 中分布式鎖主要就是靠創建臨時的順序節點來實現的。至于為什么要用順序節點和為什么用臨時節點不用持久節點？先考慮一下，下文將作出說明。

同時還有 ZK 中如何查看節點？ZK 中 ls [-w] path 為查看節點命令，-w 為添加一個 watch（監視器），/ 為查看根節點所有節點，可以看到我們剛才所創建的節點，同時如果是跟著指定節點名字的話為查看指定節點下的子節點。

后面的 00000000 為 ZK 為順序節點增加的順序。注冊監聽器也是 ZK 實現分布式鎖中比較重要的一個東西。

下面來看一下 ZK 實現分布式鎖的主要流程：

• 當第一個線程進來時會去父節點上創建一個臨時的順序節點。

• 第二個線程進來發現鎖已經被持有了，就會為當前持有鎖的節點注冊一個 watcher 監聽器。

• 第三個線程進來發現鎖已經被持有了，因為是順序節點的緣故，就會為上一個節點去創建一個 watcher 監聽器。

• 當第一個線程釋放鎖后，刪除節點，由它的下一個節點去占有鎖。

看到這里，聰明的小伙伴們都已經看出來順序節點的好處了。非順序節點的話，每進來一個線程進來都會去持有鎖的節點上注冊一個監聽器，容易引發“羊群效應”。

這么大一群羊一起向你飛奔而來，不管你頂不頂得住，反正 ZK 服務器是會增大宕機的風險。

而順序節點的話就不會，順序節點當發現已經有線程持有鎖后，會向它的上一個節點注冊一個監聽器，這樣當持有鎖的節點釋放后，也只有持有鎖的下一個節點可以搶到鎖，相當于是排好隊來執行的，降低服務器宕機風險。

至于為什么使用臨時節點，和 Redis 的過期時間一個道理，就算 ZK 服務器宕機，臨時節點會隨著服務器的宕機而消失，避免了死鎖的情況。

下面來上一段代碼的實現：

public?class?ZooKeeperDistributedLock?implements?Watcher?{private?ZooKeeper?zk;private?String?locksRoot?=?"/locks";private?String?productId;private?String?waitNode;private?String?lockNode;private?CountDownLatch?latch;private?CountDownLatch?connectedLatch?=?new?CountDownLatch(1);private?int?sessionTimeout?=?30000;public?ZooKeeperDistributedLock(String?productId)?{this.productId?=?productId;try?{String?address?=?"192.168.189.131:2181,192.168.189.132:2181";zk?=?new?ZooKeeper(address,?sessionTimeout,?this);connectedLatch.await();}?catch?(IOException?e)?{throw?new?LockException(e);}?catch?(KeeperException?e)?{throw?new?LockException(e);}?catch?(InterruptedException?e)?{throw?new?LockException(e);}}public?void?process(WatchedEvent?event)?{if?(event.getState()?==?KeeperState.SyncConnected)?{connectedLatch.countDown();return;}if?(this.latch?!=?null)?{this.latch.countDown();}}public?void?acquireDistributedLock()?{try?{if?(this.tryLock())?{return;}?else?{waitForLock(waitNode,?sessionTimeout);}}?catch?(KeeperException?e)?{throw?new?LockException(e);}?catch?(InterruptedException?e)?{throw?new?LockException(e);}}//獲取鎖public?boolean?tryLock()?{try?{//?傳入進去的locksRoot?+?“/”?+?productId//?假設productId代表了一個商品id，比如說1//?locksRoot?=?locks//?/locks/10000000000，/locks/10000000001，/locks/10000000002lockNode?=?zk.create(locksRoot?+?"/"?+?productId,?new?byte[0],?ZooDefs.Ids.OPEN_ACL_UNSAFE,?CreateMode.EPHEMERAL_SEQUENTIAL);//?看看剛創建的節點是不是最小的節點// locks：10000000000，10000000001，10000000002List<String>?locks?=?zk.getChildren(locksRoot,?false);Collections.sort(locks);if(lockNode.equals(locksRoot+"/"+?locks.get(0))){//如果是最小的節點,則表示取得鎖return?true;}//如果不是最小的節點，找到比自己小1的節點int?previousLockIndex?=?-1;for(int?i?=?0;?i?<?locks.size();?i++)?{if(lockNode.equals(locksRoot?+?“/”?+?locks.get(i)))?{previousLockIndex?=?i?-?1;break;}}this.waitNode?=?locks.get(previousLockIndex);}?catch?(KeeperException?e)?{throw?new?LockException(e);}?catch?(InterruptedException?e)?{throw?new?LockException(e);}return?false;}private?boolean?waitForLock(String?waitNode,?long?waitTime)?throws?InterruptedException,?KeeperException?{Stat?stat?=?zk.exists(locksRoot?+?"/"?+?waitNode,?true);if?(stat?!=?null)?{this.latch?=?new?CountDownLatch(1);this.latch.await(waitTime,?TimeUnit.MILLISECONDS);this.latch?=?null;}return?true;}//釋放鎖public?void?unlock()?{try?{System.out.println("unlock?"?+?lockNode);zk.delete(lockNode,?-1);lockNode?=?null;zk.close();}?catch?(InterruptedException?e)?{e.printStackTrace();}?catch?(KeeperException?e)?{e.printStackTrace();}}//異常public?class?LockException?extends?RuntimeException?{private?static?final?long?serialVersionUID?=?1L;public?LockException(String?e)?{super(e);}public?LockException(Exception?e)?{super(e);}} }

總結

既然明白了 Redis 和 ZK 分別對分布式鎖的實現，那么總該有所不同的吧。沒錯，我都幫大家整理好了：

• 實現方式的不同，Redis 實現為去插入一條占位數據，而 ZK 實現為去注冊一個臨時節點。

• 遇到宕機情況時，Redis 需要等到過期時間到了后自動釋放鎖，而 ZK 因為是臨時節點，在宕機時候已經是刪除了節點去釋放鎖。

• Redis 在沒搶占到鎖的情況下一般會去自旋獲取鎖，比較浪費性能，而 ZK 是通過注冊監聽器的方式獲取鎖，性能而言優于 Redis。

不過具體要采用哪種實現方式，還是需要具體情況具體分析，結合項目引用的技術棧來落地實現。

想知道更多？掃描下面的二維碼關注我

后臺回復"技術"，加入技術群

后臺回復“k8s”，可領取k8s資料

【精彩推薦】

• 原創|OpenAPI標準規范

• 如此簡單| ES最全詳細使用教程

• ClickHouse到底是什么？為什么如此牛逼！

• 原來ElasticSearch還可以這么理解

• 面試官：InnoDB中一棵B+樹可以存放多少行數據？

• 微服務下如何解耦？對于已經緊耦合下如何重構？

• 如何構建一套高性能、高可用、低成本的視頻處理系統？

• 架構之道：分離業務邏輯和技術細節

• 星巴克不使用兩階段提交
  

點個贊+在看，少個 bug?????

總結

以上是生活随笔為你收集整理的好文推荐 | 分布式锁用Redis好，还是Zookeeper好？的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。