分散式學習筆記七:基於zookeeper實現分散式鎖
一、分散式鎖介紹
分散式鎖主要用於在分散式環境中保護跨程序、跨主機、跨網路的共享資源實現互斥訪問,以達到保證資料的一致性。
二、架構介紹
在介紹使用Zookeeper實現分散式鎖之前,首先看當前的系統架構圖
解釋: 左邊的整個區域表示一個Zookeeper叢集,locker是Zookeeper的一個持久節點,node_1、node_2、node_3是locker這個持久節點下面的臨時順序節點。client_1、client_2、client_n表示多個客戶端,Service表示需要互斥訪問的共享資源。
三、分散式鎖獲取思路
1.獲取分散式鎖的總體思路
在獲取分散式鎖的時候在locker節點下建立臨時順序節點,釋放鎖的時候刪除該臨時節點。客戶端呼叫createNode方法在locker下建立臨時順序節點,
然後呼叫getChildren(“locker”)來獲取locker下面的所有子節點,注意此時不用設定任何Watcher。客戶端獲取到所有的子節點path之後,如果發現自己在之
前建立的子節點序號最小,那麼就認為該客戶端獲取到了鎖。如果發現自己建立的節點並非locker所有子節點中最小的,說明自己還沒有獲取到鎖,
此時客戶端需要找到比自己小的那個節點,然後對其呼叫exist()方法,同時對其註冊事件監聽器。之後,讓這個被關注的節點刪除,則客戶端的Watcher會
收到相應通知,此時再次判斷自己建立的節點是否是locker子節點中序號最小的,如皋是則獲取到了鎖,如果不是則重複以上步驟繼續獲取到比自己小的一個
節點並註冊監聽。當前這個過程中還需要許多的邏輯判斷。
2.獲取分散式鎖的核心演算法流程
下面同個一個流程圖來分析獲取分散式鎖的完整演算法,如下:
解釋:客戶端A要獲取分散式鎖的時候首先到locker下建立一個臨時順序節點(node_n),然後立即獲取locker下的所有(一級)子節點。
此時因為會有多個客戶端同一時間爭取鎖,因此locker下的子節點數量就會大於1。對於順序節點,特點是節點名稱後面自動有一個數字編號,
先建立的節點數字編號小於後建立的,因此可以將子節點按照節點名稱字尾的數字順序從小到大排序,這樣排在第一位的就是最先建立的順序節點,
此時它就代表了最先爭取到鎖的客戶端!此時判斷最小的這個節點是否為客戶端A之前創建出來的node_n,如果是則表示客戶端A獲取到了鎖,
如果不是則表示鎖已經被其它客戶端獲取,因此客戶端A要等待它釋放鎖,也就是等待獲取到鎖的那個客戶端B把自己建立的那個節點刪除。
此時就通過監聽比node_n次小的那個順序節點的刪除事件來知道客戶端B是否已經釋放了鎖,如果是,此時客戶端A再次獲取locker下的所有子節點,
再次與自己建立的node_n節點對比,直到自己建立的node_n是locker的所有子節點中順序號最小的,此時表示客戶端A獲取到了鎖!
四、基於Zookeeper的分散式鎖的程式碼實現
1.定義分散式鎖介面
定義的分散式鎖介面如下:
public interface DistributedLock {
/**獲取鎖,如果沒有得到就等待*/
public void acquire() throws Exception;
/**
* 獲取鎖,直到超時
* @param time超時時間
* @param unit time引數的單位
* @return是否獲取到鎖
* @throws Exception
*/
public boolean acquire (long time, TimeUnit unit) throws Exception;
/**
* 釋放鎖
* @throws Exception
*/
public void release() throws Exception;
}
2.定義一個簡單的互斥鎖
定義一個互斥鎖類,實現以上定義的鎖介面,同時繼承一個基類BaseDistributedLock,該基類主要用於與Zookeeper互動,
包含一個嘗試獲取鎖的方法和一個釋放鎖。
/**鎖介面的具體實現,主要藉助於繼承的父類BaseDistributedLock來實現的介面方法
* 該父類是基於Zookeeper實現分散式鎖的具體細節實現*/
public class SimpleDistributedLockMutex extends BaseDistributedLock implements DistributedLock {
/*用於儲存Zookeeper中實現分散式鎖的節點,如名稱為locker:/locker,
*該節點應該是持久節點,在該節點下面建立臨時順序節點來實現分散式鎖 */
private final String basePath;
/*鎖名稱字首,locker下建立的順序節點例如都以lock-開頭,這樣便於過濾無關節點
*這樣建立後的節點類似:lock-00000001,lock-000000002*/
private staticfinal String LOCK_NAME ="lock-";
/*用於儲存某個客戶端在locker下面建立成功的順序節點,用於後續相關操作使用(如判斷)*/
private String ourLockPath;
/**
* 用於獲取鎖資源,通過父類的獲取鎖方法來獲取鎖
* @param time獲取鎖的超時時間
* @param unit time的時間單位
* @return是否獲取到鎖
* @throws Exception
*/
private boolean internalLock (long time, TimeUnit unit) throws Exception {
//如果ourLockPath不為空則認為獲取到了鎖,具體實現細節見attemptLock的實現
ourLockPath = attemptLock(time, unit);
return ourLockPath !=null;
}
/**
* 傳入Zookeeper客戶端連線物件,和basePath
* @param client Zookeeper客戶端連線物件
* @param basePath basePath是一個持久節點
*/
public SimpleDistributedLockMutex(ZkClientExt client, String basePath){
/*呼叫父類的構造方法在Zookeeper中建立basePath節點,並且為basePath節點子節點設定字首
*同時儲存basePath的引用給當前類屬性*/
super(client,basePath,LOCK_NAME);
this.basePath = basePath;
}
/**獲取鎖,直到超時,超時後丟擲異常*/
public void acquire() throws Exception {
//-1表示不設定超時時間,超時由Zookeeper決定
if (!internalLock(-1,null)){
throw new IOException("連線丟失!在路徑:'"+basePath+"'下不能獲取鎖!");
}
}
/**
* 獲取鎖,帶有超時時間
*/
public boolean acquire(long time, TimeUnit unit) throws Exception {
return internalLock(time, unit);
}
/**釋放鎖*/
public void release()throws Exception {
releaseLock(ourLockPath);
}
}
3. 分散式鎖的實現細節
獲取分散式鎖的重點邏輯在於BaseDistributedLock,實現了基於Zookeeper實現分散式鎖的細節。
public class BaseDistributedLock {
private final ZkClientExt client;
private final String path;
private final String basePath;
private final String lockName;
private static final Integer MAX_RETRY_COUNT = 10;
public BaseDistributedLock(ZkClientExt client, String path, String lockName){
this.client = client;
this.basePath = path;
this.path = path.concat("/").concat(lockName);
this.lockName = lockName;
}
private void deleteOurPath(String ourPath) throws Exception{
client.delete(ourPath);
}
private String createLockNode(ZkClient client, String path) throws Exception{
return client.createEphemeralSequential(path, null);
}
/**
* 獲取鎖的核心方法
* @param startMillis
* @param millisToWait
* @param ourPath
* @return
* @throws Exception
*/
private boolean waitToLock(long startMillis, Long millisToWait, String ourPath) throws Exception{
boolean haveTheLock = false;
boolean doDelete = false;
try{
while ( !haveTheLock ) {
//該方法實現獲取locker節點下的所有順序節點,並且從小到大排序
List<String> children = getSortedChildren();
String sequenceNodeName = ourPath.substring(basePath.length()+1);
//計算剛才客戶端建立的順序節點在locker的所有子節點中排序位置,如果是排序為0,則表示獲取到了鎖
int ourIndex = children.indexOf(sequenceNodeName);
/*如果在getSortedChildren中沒有找到之前建立的[臨時]順序節點,這表示可能由於網路閃斷而導致
*Zookeeper認為連線斷開而刪除了我們建立的節點,此時需要丟擲異常,讓上一級去處理
*上一級的做法是捕獲該異常,並且執行重試指定的次數 見後面的 attemptLock方法 */
if ( ourIndex<0 ){
throw new ZkNoNodeException("節點沒有找到: " + sequenceNodeName);
}
//如果當前客戶端建立的節點在locker子節點列表中位置大於0,表示其它客戶端已經獲取了鎖
//此時當前客戶端需要等待其它客戶端釋放鎖,
boolean isGetTheLock = ourIndex == 0;
//如何判斷其它客戶端是否已經釋放了鎖?從子節點列表中獲取到比自己次小的哪個節點,並對其建立監聽
String pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1);
if ( isGetTheLock ){
haveTheLock = true;
}else{
//如果次小的節點被刪除了,則表示當前客戶端的節點應該是最小的了,所以使用CountDownLatch來實現等待
String previousSequencePath = basePath .concat( "/" ) .concat( pathToWatch );
final CountDownLatch latch = new CountDownLatch(1);
final IZkDataListener previousListener = new IZkDataListener() {
//次小節點刪除事件發生時,讓countDownLatch結束等待
//此時還需要重新讓程式回到while,重新判斷一次!
public void handleDataDeleted(String dataPath) throws Exception {
latch.countDown();
}
public void handleDataChange(String dataPath, Object data) throws Exception {
// ignore
}
};
try{
//如果節點不存在會出現異常
client.subscribeDataChanges(previousSequencePath, previousListener);
if ( millisToWait != null ){
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
if ( millisToWait <= 0 ){
doDelete = true; // timed out - delete our node
break;
}
latch.await(millisToWait, TimeUnit.MICROSECONDS);
}else{
latch.await();
}
}catch ( ZkNoNodeException e ){
//ignore
}finally{
client.unsubscribeDataChanges(previousSequencePath, previousListener);
}
}
}
}catch ( Exception e ){
//發生異常需要刪除節點
doDelete = true;
throw e;
}finally{
//如果需要刪除節點
if ( doDelete ){
deleteOurPath(ourPath);
}
}
return haveTheLock;
}
private String getLockNodeNumber(String str, String lockName) {
int index = str.lastIndexOf(lockName);
if ( index >= 0 ){
index += lockName.length();
return index <= str.length() ? str.substring(index) : "";
}
return str;
}
private List<String> getSortedChildren() throws Exception {
try{
List<String> children = client.getChildren(basePath);
Collections.sort(
children,
new Comparator<String>(){
public int compare(String lhs, String rhs){
return getLockNodeNumber(lhs, lockName).compareTo(getLockNodeNumber(rhs, lockName));
}
}
);
return children;
}catch(ZkNoNodeException e){
client.createPersistent(basePath, true);
return getSortedChildren();
}
}
protected void releaseLock(String lockPath) throws Exception{
deleteOurPath(lockPath);
}
protected String attemptLock(long time, TimeUnit unit) throws Exception{
final long startMillis = System.currentTimeMillis();
final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;
String ourPath = null;
boolean hasTheLock = false;
boolean isDone = false;
int retryCount = 0;
//網路閃斷需要重試一試
while ( !isDone ){
isDone = true;
try{
//createLockNode用於在locker(basePath持久節點)下建立客戶端要獲取鎖的[臨時]順序節點
ourPath = createLockNode(client, path);
/**
* 該方法用於判斷自己是否獲取到了鎖,即自己建立的順序節點在locker的所有子節點中是否最小
* 如果沒有獲取到鎖,則等待其它客戶端鎖的釋放,並且稍後重試直到獲取到鎖或者超時
*/
hasTheLock = waitToLock(startMillis, millisToWait, ourPath);
}catch ( ZkNoNodeException e ){
if ( retryCount++ < MAX_RETRY_COUNT ){
isDone = false;
}else{
throw e;
}
}
}
if ( hasTheLock ){
return ourPath;
}
return null;
}