因為每次在創建鎖和釋放鎖的過程中，都要動態創建、銷毀臨時節點來實現鎖功能。ZK中創建和刪除節點只能通過Leader服務器來執行，然后Leader服務器還需要將數據同步到所有的Follower機器上，這樣頻繁的網絡通信，性能的短板是非常突出的。在高性能，高并發的場景下，不建議使用ZooKeeper的分布式鎖。
1. Zookeeper概述
Zookeeper（后續簡稱ZK）是一個分布式的，開放源碼的分布式應用程序協調服務，通常以集群模式運轉，其協調能力可以理解為是基于觀察者設計模式來實現的；ZK服務會使用Znode存儲使用者的數據，并將這些數據以樹形目錄的形式來組織管理，支持使用者以觀察者的角色指定自己關注哪些節點\數據的變更，當這些變更發生時，ZK會通知其觀察者；為滿足本篇目標所需，著重介紹以下幾個關鍵特性：
- 數據組織：數據節點以樹形目錄(類似文件系統)組織管理，每一個節點中都會保存數據信息和節點信息。
ZooKeeper's Hierarchical Namespace
- 集群模式：通常是由3、5個基數實例組成集群，當超過半數服務實例正常工作就能對外提供服務，既能避免單點故障，又盡量高可用，每個服務實例都有一個數據備份，以實現數據全局一致
ZooKeeper Service
- 順序更新：更新請求都會轉由leader執行，來自同一客戶端的更新將按照發送的順序被寫入到ZK，處理寫請求創建Znode時，Znode名稱后會被分配一個全局唯一的遞增編號，可以通過順序號推斷請求的順序，利用這個特性可以實現高級協調服務
監聽機制：給某個節點注冊監聽器，該節點一旦發生變更（例如更新或者刪除），監聽者就會收到一個Watch Event，可以感知到節點\數據的變更
臨時節點：session鏈接斷開臨時節點就沒了，不能創建子節點（很關鍵）
ZK的分布式鎖正是基于以上特性來實現的，簡單來說是：
- 臨時節點：用于支撐異常情況下的鎖自動釋放能力
- 順序節點：用于支撐公平鎖獲取鎖和排隊等待的能力
- 監聽機制：用于支撐搶鎖能力
- 集群模式：用于支撐鎖服務的高可用
2. 加解鎖的流程描述
- 創建一個永久節點作為鎖節點（/lock2）
- 試圖加鎖的客戶端在指定鎖名稱節點（/lock2）下，創建臨時順序子節點
- 獲取鎖節點（/lock2）下所有子節點
- 對所獲取的子節點按節點自增序號從小到大排序
- 判斷自己是不是第一個子節點，若是，則獲取鎖
- 若不是，則監聽比該節點小的那個節點的刪除事件（這種只監聽前一個節點的方式避免了驚群效應）
- 若是阻塞申請鎖，則申請鎖的操作可增加阻塞等待
- 若監聽事件生效（說明前節點釋放了，可以嘗試去獲取鎖），則回到第3步重新進行判斷，直到獲取到鎖
- 解鎖時，將第一個子節點刪除釋放
3. ZK分布式鎖的能力
可能讀者是單篇閱讀，這里引入上一篇《分布式鎖上-初探》中的一些內容，一個分布式鎖應具備這樣一些功能特點：
- 互斥性：在同一時刻，只有一個客戶端能持有鎖
- 安全性：避免死鎖，如果某個客戶端獲得鎖之后處理時間超過最大約定時間，或者持鎖期間發生了故障導致無法主動釋放鎖，其持有的鎖也能夠被其他機制正確釋放，并保證后續其它客戶端也能加鎖，整個處理流程繼續正常執行
- 可用性：也被稱作容錯性，分布式鎖需要有高可用能力，避免單點故障，當提供鎖的服務節點故障（宕機）時不影響服務運行，這里有兩種模式：一種是分布式鎖服務自身具備集群模式，遇到故障能自動切換恢復工作；另一種是客戶端向多個獨立的鎖服務發起請求，當某個鎖服務故障時仍然可以從其他鎖服務讀取到鎖信息(Redlock)
- 可重入性：對同一個鎖，加鎖和解鎖必須是同一個線程，即不能把其他線程程持有的鎖給釋放了
- 高效靈活：加鎖、解鎖的速度要快；支持阻塞和非阻塞；支持公平鎖和非公平鎖
基于上文的內容，這里簡單總結一下ZK的能力矩陣（其它分布式鎖的情況會在后續文章中補充）：
能力 | ZK | MySql | Redis原生 | Redlock | ETCD |
互斥 | 是 |
|
|
|
|
安全 | 鏈接異常，session關閉后鎖會自動釋放 |
|
|
|
|
可用性 | 相對還好 |
|
|
|
|
可重入 | 線程可重入 |
|
|
|
|
加解鎖速度 | 居中 |
|
|
|
|
阻塞非阻塞 | 都支持 |
|
|
|
|
公平非公平 | 僅公平鎖 |
|
|
|
|
關于性能不太高的一種說法
因為每次在創建鎖和釋放鎖的過程中，都要動態創建、銷毀臨時節點來實現鎖功能。ZK中創建和刪除節點只能通過Leader服務器來執行，然后Leader服務器還需要將數據同步到所有的Follower機器上，這樣頻繁的網絡通信，性能的短板是非常突出的。在高性能，高并發的場景下，不建議使用ZooKeeper的分布式鎖。
由于ZooKeeper的高可用特性，在并發量不是太高的場景，也推薦使用ZK的分布式鎖。
4. InterProcessMutex 使用示例
Zookeeper 客戶端框架 Curator 提供的 InterProcessMutex 是分布式鎖的一種實現，acquire 方法阻塞|非阻塞獲取鎖，release 方法釋放鎖，另外還提供了可撤銷、可重入功能。
4.1 接口介紹
// 獲取互斥鎖
public void acquire() throws Exception;
// 在給定的時間內獲取互斥鎖
public boolean acquire(long time, TimeUnit unit) throws Exception;
// 釋放鎖處理
public void release() throws Exception;
// 如果當前線程獲取了互斥鎖，則返回true
boolean isAcquiredInThisProcess();
4.2 pom依賴
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8.2</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.5.7</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>4.3.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.3.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-client</artifactId>
<version>4.3.0</version>
</dependency>
4.3 示例
package com.atguigu.case3;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class CuratorLockTest {
public static void main(String[] args) {
// 創建分布式鎖1
InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");
// 創建分布式鎖2
InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");
new Thread(new Runnable() {
@Override
public void run() {
try {
lock1.acquire();
System.out.println("線程1 獲取到鎖");
lock1.acquire();
System.out.println("線程1 再次獲取到鎖");
Thread.sleep(5 * 1000);
lock1.release();
System.out.println("線程1 釋放鎖");
lock1.release();
System.out.println("線程1 再次釋放鎖");
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
try {
lock2.acquire();
System.out.println("線程2 獲取到鎖");
lock2.acquire();
System.out.println("線程2 再次獲取到鎖");
Thread.sleep(5 * 1000);
lock2.release();
System.out.println("線程2 釋放鎖");
lock2.release();
System.out.println("線程2 再次釋放鎖");
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
}
private static CuratorFramework getCuratorFramework() {
ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
CuratorFramework client = CuratorFrameworkFactory.builder().connectString("xxx:2181,xxx:2181,xxx:2181")
.connectionTimeoutMs(2000)
.sessionTimeoutMs(2000)
.retryPolicy(policy).build();
// 啟動客戶端
client.start();
System.out.println("zookeeper 啟動成功");
return client;
}
}
5. DIY一個閹割版的分布式鎖
通過這個實例對照第2節內容來理解加解鎖的流程，以及如何避免驚群效應。
package com.rock.case2;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;
/**
* zk 分布式鎖 v1版本:
* 完成功能 :
* 1. 避免了驚群效應
* 缺失功能:
* 1. 超時控制
* 2. 讀寫鎖
* 3. 重入控制
*/
public class DistributedLock {
private String connectString;
private int sessionTimeout;
private ZooKeeper zk;
private CountDownLatch connectLatch = new CountDownLatch(1);
private CountDownLatch waitLatch = new CountDownLatch(1);
private String waitPath;
private String currentNode;
private String LOCK_ROOT_PATH;
private static String NODE_PREFIX = "w";
public DistributedLock(String connectString, int sessionTimeout, String lockName){
//TODO:數據校驗
this.connectString = connectString;
this.sessionTimeout = sessionTimeout;
this.LOCK_ROOT_PATH = lockName;
}
public void init() throws IOException, KeeperException, InterruptedException {
// 建聯
zk = new ZooKeeper(connectString, sessionTimeout, watchedEvent -> {
// connectLatch 連接上zk后 釋放
if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
connectLatch.countDown();
}
});
connectLatch.await();// 等待zk正常連接后
// 判斷鎖名稱節點是否存在
Stat stat = zk.exists(LOCK_ROOT_PATH, false);
if (stat == null) {
// 創建一下鎖名稱節點
try {
zk.create(LOCK_ROOT_PATH, LOCK_ROOT_PATH.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} catch (KeeperException e) {
//并發創建沖突忽略。
if (!e.code().name().equals("NODEEXISTS")) {
throw e;
}
}
}
}
/**
* 待補充功能:
* 1. 超時設置
* 2. 讀寫區分
* 3. 重入控制
*/
public void zklock() throws KeeperException, InterruptedException {
if (!tryLock()) {
waitLock();
zklock();
}
}
/**
*
*/
private void waitLock() throws KeeperException, InterruptedException {
try {
zk.getData(waitPath, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent){
// waitLatch 需要釋放
if (watchedEvent.getType() == Watcher.Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
waitLatch.countDown();
}
}
}, new Stat());
// 等待監聽
waitLatch.await();
} catch (KeeperException.NoNodeException e) {
//如果等待的節點已經被清除了,不等了,再嘗試去搶鎖
return;
}
}
private boolean tryLock() throws KeeperException, InterruptedException {
currentNode = zk.create(LOCK_ROOT_PATH + "/" + NODE_PREFIX, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
// 判斷創建的節點是否是最小的序號節點，如果是獲取到鎖；如果不是，監聽他序號前一個節點
List<String> children = zk.getChildren(LOCK_ROOT_PATH, false);
// 如果children 只有一個值，那就直接獲取鎖； 如果有多個節點，需要判斷，誰最小
if (children.size() == 1) {
return true;
} else {
String thisNode = currentNode.substring(LOCK_ROOT_PATH.length() + 1);
// 通過w00000000獲取該節點在children集合的位置
int index = children.indexOf(thisNode);
if (index == 0) {
//自己就是第一個節點
return true;
}
// 需要監聽 他前一個節點變化
waitPath = LOCK_ROOT_PATH + "/" + children.get(index - 1);
}
return false;
}
// 解鎖
public void unZkLock(){
// 刪除節點
try {
zk.delete(this.currentNode, -1);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
}
本文轉載自微信公眾號「架構染色」，可以通過以下二維碼關注。轉載本文請聯系【架構染色】公眾號作者。
