ZooKeeper 分散式鎖

摘要： 在Redis分散式鎖一文中， 作者介紹瞭如何使用Redis開發分散式鎖。 Redis分散式鎖具有輕量高吞吐量的特點，但是一致性保證較弱。我們可以使用Zookeeper開發分散式鎖，來滿足對高一致性的要求。 Zookeeper 分散式鎖原理 Zookeeper 節點具有一些性質...

在Redis分散式鎖一文中， 作者介紹瞭如何使用Redis開發分散式鎖。

Redis分散式鎖具有輕量高吞吐量的特點，但是一致性保證較弱。我們可以使用Zookeeper開發分散式鎖，來滿足對高一致性的要求。

Zookeeper 分散式鎖原理

Zookeeper 節點具有一些性質可以幫助我們開發分散式鎖:

• 臨時節點: 客戶端可以建立臨時節點，當客戶端會話終止或超時後Zookeeper會自動刪除臨時節點。該特性可以用來避免死鎖。
• 觸發器: 當節點的狀態發生改變時，Zookeeper會通知監聽相應事件的客戶端。該特性可以用來實現阻塞等待加鎖。
• 有序節點: 客戶端可以在某個節點下建立子節點，Zookeeper會根據子節點數量自動生成整數序號，類似於資料庫的自增主鍵。

一種比較容易想到的分散式鎖實現方案是:

1. 檢查鎖節點是否已經建立，若未建立則嘗試建立一個臨時節點
2. 若臨時節點建立成功說明已成功加鎖。若持有鎖的客戶端崩潰或網路異常無法維持Session，鎖節點會被刪除不會產生死鎖。
3. 若臨時節點建立失敗說明加鎖失敗，等待加鎖。watch鎖節點exists事件，當接收到節點被刪除的通知後再次嘗試加鎖。
4. 因為Zookeeper中的Watch是一次性的，若再次嘗試加鎖失敗，需要重新設定Watch。
5. 操作完成後，刪除鎖節點釋放鎖。

該方案存在的問題是，當鎖被釋放時Zookeeper需要通知大量訂閱了該事件的客戶端，這種現象稱為"驚群現象"或"羊群效應"。

驚群現象對Zookeeper正常提供服務非常不利，因此實踐中通常採取另一種方案:

1. 建立一個永久節點作為鎖節點，試圖加鎖的客戶端在鎖節點下建立臨時順序節點。Zookeeper會保證子節點的有序性。
2. 若鎖節點下id最小的節點是為當前客戶端建立的節點，說明當前客戶端成功加鎖。
3. 否則加鎖失敗，訂閱上一個順序節點。當上一個節點被刪除時，當前節點為最小，說明加鎖成功。
4. 操作完成後，刪除鎖節點釋放鎖。

該方案每次鎖釋放時只需要通知一個客戶端，避免驚群現象發生。

該方案的特徵是優先排隊等待的客戶端會先獲得鎖，這種鎖稱為公平鎖。而鎖釋放後，所有客戶端重新競爭鎖的方案稱為非公平鎖。

Demo

本節作者將使用Zookeeper官方Java API實現一個簡單的公平鎖。

使用Maven進行依賴管理，專案依賴 Zookeeper 官方 java sdk 和 apache commons-lang3工具包:

<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.6</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.5</version>
<type>pom</type>
</dependency>

點選檢視ofollow,noindex" target="_blank">完整程式碼 :

package zk;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;

/**
 * @author finley
 */
public class ZKLock {

private ZooKeeper zk;

private String basePath;

private String lockPath;

private static final byte[] LOCK_DATA = "".getBytes();

// zk 為客戶端連線例項, basePath 為鎖節點路徑，我們將在 basePath 下建立順序子節點
public ZKLock(ZooKeeper zk, String basePath) {
// 按照 zk 的路徑規則，以'/'開始，不得以'/'結束
if (basePath.endsWith("/") || !basePath.startsWith("/")) {
throw new IllegalArgumentException("base path must start with '/', and must not end with '/'");
}
this.zk = zk;
this.basePath = basePath;
}

// 檢測 basePath 節點是否存在, 若不存在則建立
private void ensureBasePath() throws KeeperException, InterruptedException {
if (zk.exists(basePath, false) == null) {
// basePath 不存在，進行建立
List<String> pathParts = new ArrayList<>(Arrays.asList(basePath.split("/"))); // 將路徑處理為節點列表
pathParts.remove(0); //因為 basePath 以'/'開始, pathParts[0] 一定是空串，將其移除

// 自底向上，尋找路徑中最後一個存在的節點
int last = 0;
for (int i = pathParts.size() - 1; i >= 0; i--) {
String path = "/" + StringUtils.join(pathParts.subList(0, i), '/');
if (zk.exists(path, false) != null) {
last = i;
break;
}
}

// 從最後一個存在的節點開始，依次建立節點
for (int i = last; i < pathParts.size(); i++) {
String path = "/" + StringUtils.join(pathParts.subList(0, i + 1), '/');
try {
zk.create(path, LOCK_DATA, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} catch (KeeperException.NodeExistsException ignore) {} // may created by other thread
}

}
}

// 阻塞直至加鎖成功
public void lock() throws KeeperException, InterruptedException {
ensureBasePath();

// 在 basePath 下建立臨時順序子節點
String lockPath = zk.create(basePath + "/lock_", LOCK_DATA, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

System.out.println(Thread.currentThread().getName() + " create: " + lockPath);

// 迴圈檢查加鎖是否成功
while(true) {
// 取出 basePath 中所有節點並找到最小子節點
// 因為順序子節點總是遞增的，新建立的節點一定比當前 lockPath 更大，所以 create 和 getChildren 兩個操作不保持原子性不會出現異常
List<String> children = zk.getChildren(basePath,false);
Collections.sort(children);
String minNode = children.get(0);

// 當前執行緒建立了最小子節點，加鎖成功
if (StringUtils.isNotBlank(lockPath) && StringUtils.isNotBlank(minNode) && StringUtils.equals(lockPath, basePath + "/" + minNode) {
this.lockPath = lockPath; // 加鎖成功，寫入鎖路徑
return;
}

// 加鎖失敗，設定 watch
String watchNode = null;
String node = lockPath.substring(lockPath.lastIndexOf("/") + 1);
for (int i = children.size() - 1; i >= 0; i--) {
String child = children.get(i);
if (child.compareTo(node) < 0) {
watchNode = child;
break;
}
}

// 找到需要監視的節點，設定 watch
if (watchNode != null) {
System.out.println(Thread.currentThread().getName() + " watch: " + watchNode);

String watchPath = basePath + "/" + watchNode;

// 監視 getData 而非 exists 的原因是: 在獲取子節點和設定 watch 這段時間內，被監視的節點可能已被刪除(鎖釋放/持有者崩潰)
// exists 監視會成功設定，但永遠不會觸發NodeDeleted事件(順序子節點序號自增，不會複用使用過的序號)。本方法會無限制等待下去
// 若被監視節點已刪除，getData 會丟擲異常，避免執行緒浪費時間等待

// 該呼叫中的 watch 回調當事件發生時會在另一個執行緒中執行
try {
zk.getData(watchPath, event -> {
if(event.getType() == Watcher.Event.EventType.NodeDeleted) {
// 主執行緒會呼叫 this.wait()
// fixme: 這裡有一個bug，若事件型別不是 NodeDeleted 應進行處理。分散式鎖不會產生這種情況，可能是其它客戶端操作所致
synchronized (this) {
notifyAll();
}
}
}, null);
} catch(KeeperException.NoNodeException e) {
// 因為上一個節點被刪除導致 getData watch 失敗，進入下一個次迴圈，重新檢查自己是否已持有鎖
continue;
}


synchronized (this) {
// 等待被 watch 喚醒，喚醒後進入下一次迴圈，重新檢查確認自己已持有鎖
wait();
System.out.println(Thread.currentThread().getName() + " notified");
}
}
}

}

// 釋放鎖
public void unlock() throws KeeperException, InterruptedException {
// 加鎖成功時會將鎖路徑寫入 lockPath
if (StringUtils.isNotBlank(lockPath)) {
zk.delete(lockPath, -1); // 刪除鎖記錄釋放鎖
} else {
throw new IllegalStateException("don't has lock"); // 未設定鎖記錄說明本執行緒未持有鎖
}
}

public static void main(String[] args) {
int concurrent = 10;
ExecutorService service = Executors.newFixedThreadPool(concurrent);
for (int i = 0; i < concurrent; i++) {
service.execute(() -> {
// 為保證各執行緒獨立的持有鎖，每個執行緒應持有獨立的 zookeeper 會話
ZooKeeper zk;
try {

zk = new ZooKeeper("localhost:2181", 6000, watchedEvent -> {
if (Watcher.Event.KeeperState.SyncConnected == watchedEvent.getState())
System.out.println("connection is established...");
});

ZKLock lock = new ZKLock(zk, "/test/node1");

lock.lock();
System.out.println(Thread.currentThread().getName() + "acquire success");

Thread.sleep(1000);
System.out.println("do sth, thread: " + Thread.currentThread().getName());

lock.unlock();
System.out.println(Thread.currentThread().getName() + "release success");

} catch (Exception e) {
e.printStackTrace();
}
});
}
service.shutdown();
}

}

Curator

Cruator 是一個 Zookeeper 工具集, 提供了包括分散式鎖在內的常用應用的封裝，本文以 Cruator 的分散式鎖實現原始碼為例進行分析。

使用maven安裝依賴:

<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.0.1</version>
</dependency>

編寫加鎖程式碼:

public class ZkLock {

public static void main(String[] args) throws Exception {
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181", retryPolicy);
client.start();

// 鎖節點為 /curator/mutex
InterProcessMutex mutex = new InterProcessMutex(client, "/curator/mutex");
try {
// 嘗試加鎖
mutex.acquire();
// 完成業務
System.out.println("foo bar");
} finally {
// 釋放鎖
mutex.release();
client.close();
}

}

}

接下來分析InterProcessMutex.acquire() 的實現:

/**
 * Acquire the mutex - blocking until it's available. Note: the same thread
 * can call acquire re-entrantly. Each call to acquire must be balanced by a call
 * to {@link #release()}
 *
 * @throws Exception ZK errors, connection interruptions
*/
@Override
public void acquire() throws Exception
{
if ( !internalLock(-1, null) )
{
throw new IOException("Lost connection while trying to acquire lock: " + basePath);
}
}

接下來看internalLock 方法:

private boolean internalLock(long time, TimeUnit unit) throws Exception
{
Thread currentThread = Thread.currentThread();

// threadData 是一個 ConcurrentMap, 記錄各執行緒鎖的狀態
LockData lockData = threadData.get(currentThread);
if ( lockData != null ) // lockData 不為空， 說明執行緒已經持有鎖
{
// 重入鎖，重入計數器增加
lockData.lockCount.incrementAndGet();
return true;
}

// internals.attemptLock 完成實際的訪問Zookeeper獲取鎖的操作
String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
if ( lockPath != null )
{
LockData newLockData = new LockData(currentThread, lockPath);
threadData.put(currentThread, newLockData);
return true;
}

return false;
}

分析實際執行加鎖操作的internals.attemptLock 方法:

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
{
final longstartMillis = System.currentTimeMillis();
final LongmillisToWait = (unit != null) ? unit.toMillis(time) : null;
final byte[]localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
intretryCount = 0;

StringourPath = null;
booleanhasTheLock = false;
booleanisDone = false;

// 自旋加鎖
while ( !isDone )
{
isDone = true;

try
{
// 在鎖節點下建立臨時順序節點
ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
// 等待自己的節點成為最小的節點，即加鎖成功
hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
}
catch ( KeeperException.NoNodeException e )
{
// 當 session 超時會丟擲異常，根據重試策略直接進行重試 
if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
{
isDone = false;
}
else
{
throw e;
}
}
}

if ( hasTheLock )
{
return ourPath;
}

return null;
}

首先閱讀StandardLockInternalsDriver.createsTheLock() 原始碼:

public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception
{
String ourPath;
if ( lockNodeBytes != null )
{
ourPath = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, lockNodeBytes);
}
else
{
ourPath = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
}
return ourPath;
}

建立臨時順序節點, 不再贅述。

接下來檢視internalLockLoop :

while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
{
// 獲得所有子節點，按序號升序排列
List<String>children = getSortedChildren();

// 判斷自己是否為序號最小的節點
StringsequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
PredicateResultspredicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
if ( predicateResults.getsTheLock() )
{
haveTheLock = true;
}
else
{
// 獲得前一個節點的路徑
StringpreviousSequencePath = basePath + "/" + predicateResults.getPathToWatch();

// 監聽前一個節點並進行wait(), 當鎖被釋放時會通過notifyall() 喚醒
synchronized(this)
{
try 
{
// 使用getData()而非exists()監聽器的原因是:
// 若此時前一個節點已被刪除exists()仍會成功設定，但不可能被觸發(順序節點不會再次使用前一個節點的序號)。這會使方法浪費時間等待，也屬於Zookeeper資源浪費
// 若前一個節點被刪除getData() 會丟擲異常
client.getData().usingWatcher(watcher).forPath(previousSequencePath);

// 若設定了等待時間
if ( millisToWait != null )
{
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
if ( millisToWait <= 0 )
{
doDelete = true;// timed out - delete our node
break;
}
// 等待指定的時間
wait(millisToWait);
}
else
{
// 永遠等待
wait();
}
}
catch ( KeeperException.NoNodeException e ) 
{
// getData() 丟擲此異常說明前一個節點已被刪除, 重新嘗試獲取鎖。
}
}
}
}