提到鎖大家肯定有了解，像 Synchronized、ReentrantLock，在單程序情況下，多個執行緒訪問同一資源，可以用它們來保證執行緒的安全性。
圖片來自 Pexels
不過目前網際網路專案越來越多的專案採用叢集部署，也就是分散式情況，這兩種鎖就有些不夠用了。
來兩張圖舉例說明下，本地鎖的情況下：
分散式鎖情況下：
就其思想來說，就是一種“我全都要”的思想，所有服務都到一個統一的地方來取鎖，只有取到鎖的才能繼續執行下去。
說完思想，下面來說一下具體的實現。
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 分鐘了，就直接把門給撬開了，一看裡面卻是張四，這就很尷尬啊。
多說無益，煩人，直接上程式碼：
//基於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; // 隨機生成一個 value String 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，不用擔心業務執行時間超長而鎖被自動刪掉。
前面也提到了鎖的自動續期，我們來看看 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/10000000002 lockNode = zk.create(locksRoot + "/" + productId, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); // 看看剛建立的節點是不是最小的節點 // locks：10000000000，10000000001，10000000002 List<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。
不過具體要採用哪種實現方式，還是需要具體情況具體分析，結合專案引用的技術棧來落地實現。
出處：juejin.im/post/6891571079702118407