基於zookeeper的分散式鎖
實現分散式鎖目前有三種流行方案，分別為基於資料庫、Redis、Zookeeper的方案，其中前兩種方案網路上有很多資料可以參考，本文不做展開。我們來看下使用Zookeeper如何實現分散式鎖。
什麼是Zookeeper？
Zookeeper（業界簡稱zk）是一種提供配置管理、分散式協同以及命名的中心化服務，這些提供的功能都是分散式系統中非常底層且必不可少的基本功能，但是如果自己實現這些功能而且要達到高吞吐、低延遲同時還要保持一致性和可用性，實際上非常困難。因此zookeeper提供了這些功能，開發者在zookeeper之上構建自己的各種分散式系統。
雖然zookeeper的實現比較複雜，但是它提供的模型抽象卻是非常簡單的。Zookeeper提供一個多層級的節點名稱空間（節點稱為znode），每個節點都用一個以斜槓（/）分隔的路徑表示，而且每個節點都有父節點（根節點除外），非常類似於檔案系統。例如，/foo/doo這個表示一個znode，它的父節點為/foo，父父節點為/，而/為根節點沒有父節點。與檔案系統不同的是，這些節點都可以設定關聯的資料，而檔案系統中只有檔案節點可以存放資料而目錄節點不行。Zookeeper為了保證高吞吐和低延遲，在記憶體中維護了這個樹狀的目錄結構，這種特性使得Zookeeper不能用於存放大量的資料，每個節點的存放資料上限為1M。
而為了保證高可用，zookeeper需要以叢集形態來部署，這樣只要叢集中大部分機器是可用的（能夠容忍一定的機器故障），那麼zookeeper本身仍然是可用的。客戶端在使用zookeeper時，需要知道叢集機器列表，通過與叢集中的某一臺機器建立TCP連線來使用服務，客戶端使用這個TCP連結來發送請求、獲取結果、獲取監聽事件以及傳送心跳包。如果這個連線異常斷開了，客戶端可以連線到另外的機器上。
架構簡圖如下所示：
zookeeper
客戶端的讀請求可以被叢集中的任意一臺機器處理，如果讀請求在節點上註冊了監聽器，這個監聽器也是由所連線的zookeeper機器來處理。對於寫請求，這些請求會同時發給其他zookeeper機器並且達成一致後，請求才會返回成功。因此，隨著zookeeper的叢集機器增多，讀請求的吞吐會提高但是寫請求的吞吐會下降。
有序性是zookeeper中非常重要的一個特性，所有的更新都是全域性有序的，每個更新都有一個唯一的時間戳，這個時間戳稱為zxid（Zookeeper Transaction Id）。而讀請求只會相對於更新有序，也就是讀請求的返回結果中會帶有這個zookeeper最新的zxid。
如何使用zookeeper實現分散式鎖？
在描述演算法流程之前，先看下zookeeper中幾個關於節點的有趣的性質：
-
有序節點：假如當前有一個父節點為/lock，我們可以在這個父節點下面建立子節點；zookeeper提供了一個可選的有序特性，例如我們可以建立子節點“/lock/node-”並且指明有序，那麼zookeeper在生成子節點時會根據當前的子節點數量自動新增整數序號，也就是說如果是第一個建立的子節點，那麼生成的子節點為/lock/node-0000000000，下一個節點則為/lock/node-0000000001，依次類推。
-
臨時節點：客戶端可以建立一個臨時節點，在會話結束或者會話超時後，zookeeper會自動刪除該節點。
-
事件監聽：在讀取資料時，我們可以同時對節點設定事件監聽，當節點資料或結構變化時，zookeeper會通知客戶端。當前zookeeper有如下四種事件：1）節點建立；2）節點刪除；3）節點資料修改；4）子節點變更。
下面描述使用zookeeper實現分散式鎖的演算法流程，假設鎖空間的根節點為/lock：
-
客戶端連線zookeeper，並在/lock下建立臨時的且有序的子節點，第一個客戶端對應的子節點為/lock/lock-0000000000，第二個為/lock/lock-0000000001，以此類推。
-
客戶端獲取/lock下的子節點列表，判斷自己建立的子節點是否為當前子節點列表中序號最小的子節點，如果是則認為獲得鎖，否則監聽/lock的子節點變更訊息，獲得子節點變更通知後重復此步驟直至獲得鎖；
-
執行業務程式碼；
-
完成業務流程後，刪除對應的子節點釋放鎖。
步驟1中建立的臨時節點能夠保證在故障的情況下鎖也能被釋放，考慮這麼個場景：假如客戶端a當前建立的子節點為序號最小的節點，獲得鎖之後客戶端所在機器宕機了，客戶端沒有主動刪除子節點；如果建立的是永久的節點，那麼這個鎖永遠不會釋放，導致死鎖；由於建立的是臨時節點，客戶端宕機後，過了一定時間zookeeper沒有收到客戶端的心跳包判斷會話失效，將臨時節點刪除從而釋放鎖。
另外細心的朋友可能會想到，在步驟2中獲取子節點列表與設定監聽這兩步操作的原子性問題，考慮這麼個場景：客戶端a對應子節點為/lock/lock-0000000000，客戶端b對應子節點為/lock/lock-0000000001，客戶端b獲取子節點列表時發現自己不是序號最小的，但是在設定監聽器前客戶端a完成業務流程刪除了子節點/lock/lock-0000000000，客戶端b設定的監聽器豈不是丟失了這個事件從而導致永遠等待了？這個問題不存在的。因為zookeeper提供的API中設定監聽器的操作與讀操作是原子執行的，也就是說在讀子節點列表時同時設定監聽器，保證不會丟失事件。
最後，對於這個演算法有個極大的優化點：假如當前有1000個節點在等待鎖，如果獲得鎖的客戶端釋放鎖時，這1000個客戶端都會被喚醒，這種情況稱為“羊群效應”；在這種羊群效應中，zookeeper需要通知1000個客戶端，這會阻塞其他的操作，最好的情況應該只喚醒新的最小節點對應的客戶端。應該怎麼做呢？在設定事件監聽時，每個客戶端應該對剛好在它之前的子節點設定事件監聽，例如子節點列表為/lock/lock-0000000000、/lock/lock-0000000001、/lock/lock-0000000002，序號為1的客戶端監聽序號為0的子節點刪除訊息，序號為2的監聽序號為1的子節點刪除訊息。
zookeeper學習中
所以調整後的分散式鎖演算法流程如下：
-
客戶端連線zookeeper，並在/lock下建立臨時的且有序的子節點，第一個客戶端對應的子節點為/lock/lock-0000000000，第二個為/lock/lock-0000000001，以此類推；
-
客戶端獲取/lock下的子節點列表，判斷自己建立的子節點是否為當前子節點列表中序號最小的子節點，如果是則認為獲得鎖，否則監聽剛好在自己之前一位的子節點刪除訊息，獲得子節點變更通知後重復此步驟直至獲得鎖；
-
執行業務程式碼；
-
完成業務流程後，刪除對應的子節點釋放鎖。
Curator的原始碼分析
雖然zookeeper原生客戶端暴露的API已經非常簡潔了，但是實現一個分散式鎖還是比較麻煩的…我們可以直接使用curator這個開源專案提供的zookeeper分散式鎖實現。
我們只需要引入下面這個包（基於maven）：
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.0.0</version>
</dependency>
然後就可以用啦！程式碼如下：
public static void main(String[] args) throws Exception {
//建立zookeeper的客戶端
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
CuratorFramework client = CuratorFrameworkFactory.newClient("10.21.41.181:2181,10.21.42.47:2181,10.21.49.252:2181", retryPolicy);
client.start();
//建立分散式鎖, 鎖空間的根節點路徑為/curator/lock
InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");
mutex.acquire();
//獲得了鎖, 進行業務流程
System.out.println("Enter mutex");
//完成業務流程, 釋放鎖
mutex.release();
//關閉客戶端
client.close();
}
可以看到關鍵的核心操作就只有mutex.acquire()和mutex.release()，簡直太方便了！
下面來分析下獲取鎖的原始碼實現。acquire的方法如下：
/*
* 獲取鎖，當鎖被佔用時會阻塞等待，這個操作支援同線程的可重入（也就是重複獲取鎖），acquire的次數需要與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);
}
}
這裡有個地方需要注意，當與zookeeper通訊存在異常時，acquire會直接丟擲異常，需要使用者自身做重試策略。程式碼中呼叫了internalLock(-1, null)，引數表明在鎖被佔用時永久阻塞等待。internalLock的程式碼如下：
private boolean internalLock(long time, TimeUnit unit) throws Exception
{
//這裡處理同線程的可重入性，如果已經獲得鎖，那麼只是在對應的資料結構中增加acquire的次數統計，直接返回成功
Thread currentThread = Thread.currentThread();
LockData lockData = threadData.get(currentThread);
if ( lockData != null )
{
// re-entering
lockData.lockCount.incrementAndGet();
return true;
}
//這裡才真正去zookeeper中獲取鎖
String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
if ( lockPath != null )
{
//獲得鎖之後，記錄當前的執行緒獲得鎖的資訊，在重入時只需在LockData中增加次數統計即可
LockData newLockData = new LockData(currentThread, lockPath);
threadData.put(currentThread, newLockData);
return true;
}
//在阻塞返回時仍然獲取不到鎖，這裡上下文的處理隱含的意思為zookeeper通訊異常
return false;
}
程式碼中增加了具體註釋，不做展開。看下zookeeper獲取鎖的具體實現：
String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
{
//引數初始化，此處省略
//...
//自旋獲取鎖
while ( !isDone )
{
isDone = true;
try
{
//在鎖空間下建立臨時且有序的子節點
ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
//判斷是否獲得鎖（子節點序號最小），獲得鎖則直接返回，否則阻塞等待前一個子節點刪除通知
hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
}
catch ( KeeperException.NoNodeException e )
{
//對於NoNodeException，程式碼中確保了只有發生session過期才會在這裡丟擲NoNodeException，因此這裡根據重試策略進行重試
if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
{
isDone = false;
}
else
{
throw e;
}
}
}
//如果獲得鎖則返回該子節點的路徑
if ( hasTheLock )
{
return ourPath;
}
return null;
}
上面程式碼中主要有兩步操作：
-
driver.createsTheLock：建立臨時且有序的子節點，裡面實現比較簡單不做展開，主要關注幾種節點的模式：1）PERSISTENT（永久）；2）PERSISTENT_SEQUENTIAL（永久且有序）；3）EPHEMERAL（臨時）；4）EPHEMERAL_SEQUENTIAL（臨時且有序）。
-
internalLockLoop：阻塞等待直到獲得鎖。
看下internalLockLoop是怎麼判斷鎖以及阻塞等待的，這裡刪除了一些無關程式碼，只保留主流程：
//自旋直至獲得鎖
while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
{
//獲取所有的子節點列表，並且按序號從小到大排序
List<String> children = getSortedChildren();
//根據序號判斷當前子節點是否為最小子節點
String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
if ( predicateResults.getsTheLock() )
{
//如果為最小子節點則認為獲得鎖
haveTheLock = true;
}
else
{
//否則獲取前一個子節點
String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
//這裡使用物件監視器做執行緒同步，當獲取不到鎖時監聽前一個子節點刪除訊息並且進行wait()，當前一個子節點刪除（也就是鎖釋放）時，回撥會通過notifyAll喚醒此執行緒，此執行緒繼續自旋判斷是否獲得鎖
synchronized(this)
{
try
{
//這裡使用getData()介面而不是checkExists()是因為，如果前一個子節點已經被刪除了那麼會丟擲異常而且不會設定事件監聽器，而checkExists雖然也可以獲取到節點是否存在的資訊但是同時設定了監聽器，這個監聽器其實永遠不會觸發，對於zookeeper來說屬於資源洩露
client.getData().usingWatcher(watcher).forPath(previousSequencePath);
//如果設定了阻塞等待的時間
if ( millisToWait != null )
{
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
if ( millisToWait <= 0 )
{
doDelete = true; // 等待時間到達，刪除對應的子節點
break;
}
//等待相應的時間
wait(millisToWait);
}
else
{
//永遠等待
wait();
}
}
catch ( KeeperException.NoNodeException e )
{
//上面使用getData來設定監聽器時，如果前一個子節點已經被刪除那麼會丟擲NoNodeException，只需要自旋一次即可，無需額外處理
}
}
}
}
具體邏輯見註釋，不再贅述。程式碼中設定的事件監聽器，在事件發生回撥時只是簡單的notifyAll喚醒當前執行緒以重新自旋判斷，比較簡單不再展開。