ZooKeeper示例 分佈式鎖

場景描述

在分佈式應用, 每每存在多個進程提供同一服務. 這些進程有可能在相同的機器上, 也有可能分佈在不一樣的機器上. 若是這些進程共享了一些資源, 可能就須要分佈式鎖來鎖定對這些資源的訪問.
本文將介紹如何利用zookeeper實現分佈式鎖.git

思路

進程須要訪問共享數據時, 就在"/locks"節點下建立一個sequence類型的子節點, 稱爲thisPath. 當thisPath在全部子節點中最小時, 說明該進程得到了鎖. 進程得到鎖以後, 就能夠訪問共享資源了. 訪問完成後, 須要將thisPath刪除. 鎖由新的最小的子節點得到.
有了清晰的思路以後, 還須要補充一些細節. 進程如何知道thisPath是全部子節點中最小的呢? 能夠在建立的時候, 經過getChildren方法獲取子節點列表, 而後在列表中找到排名比thisPath前1位的節點, 稱爲waitPath, 而後在waitPath上註冊監聽, 當waitPath被刪除後, 進程得到通知, 此時說明該進程得到了鎖.github

實現

以一個DistributedClient對象模擬一個進程的形式, 演示zookeeper分佈式鎖的實現.web

  1. public class DistributedClient {  安全

  2.     // 超時時間  網絡

  3.     private static final int SESSION_TIMEOUT = 5000;  app

  4.     // zookeeper server列表  分佈式

  5.     private String hosts = "localhost:4180,localhost:4181,localhost:4182";  this

  6.     private String groupNode = "locks";  spa

  7.     private String subNode = "sub";  線程

  8.   

  9.     private ZooKeeper zk;  

  10.     // 當前client建立的子節點  

  11.     private String thisPath;  

  12.     // 當前client等待的子節點  

  13.     private String waitPath;  

  14.   

  15.     private CountDownLatch latch = new CountDownLatch(1);  

  16.   

  17.     /** 

  18.      * 鏈接zookeeper 

  19.      */  

  20.     public void connectZookeeper() throws Exception {  

  21.         zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {  

  22.             public void process(WatchedEvent event) {  

  23.                 try {  

  24.                     // 鏈接創建時, 打開latch, 喚醒wait在該latch上的線程  

  25.                     if (event.getState() == KeeperState.SyncConnected) {  

  26.                         latch.countDown();  

  27.                     }  

  28.   

  29.                     // 發生了waitPath的刪除事件  

  30.                     if (event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {  

  31.                         doSomething();  

  32.                     }  

  33.                 } catch (Exception e) {  

  34.                     e.printStackTrace();  

  35.                 }  

  36.             }  

  37.         });  

  38.   

  39.         // 等待鏈接創建  

  40.         latch.await();  

  41.   

  42.         // 建立子節點  

  43.         thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,  

  44.                 CreateMode.EPHEMERAL_SEQUENTIAL);  

  45.   

  46.         // wait一小會, 讓結果更清晰一些  

  47.         Thread.sleep(10);  

  48.   

  49.         // 注意, 沒有必要監聽"/locks"的子節點的變化狀況  

  50.         List<String> childrenNodes = zk.getChildren("/" + groupNode, false);  

  51.   

  52.         // 列表中只有一個子節點, 那確定就是thisPath, 說明client得到鎖  

  53.         if (childrenNodes.size() == 1) {  

  54.             doSomething();  

  55.         } else {  

  56.             String thisNode = thisPath.substring(("/" + groupNode + "/").length());  

  57.             // 排序  

  58.             Collections.sort(childrenNodes);  

  59.             int index = childrenNodes.indexOf(thisNode);  

  60.             if (index == -1) {  

  61.                 // never happened  

  62.             } else if (index == 0) {  

  63.                 // inddx == 0, 說明thisNode在列表中最小, 當前client得到鎖  

  64.                 doSomething();  

  65.             } else {  

  66.                 // 得到排名比thisPath前1位的節點  

  67.                 this.waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);  

  68.                 // 在waitPath上註冊監聽器, 當waitPath被刪除時, zookeeper會回調監聽器的process方法  

  69.                 zk.getData(waitPath, truenew Stat());  

  70.             }  

  71.         }  

  72.     }  

  73.   

  74.     private void doSomething() throws Exception {  

  75.         try {  

  76.             System.out.println("gain lock: " + thisPath);  

  77.             Thread.sleep(2000);  

  78.             // do something  

  79.         } finally {  

  80.             System.out.println("finished: " + thisPath);  

  81.             // 將thisPath刪除, 監聽thisPath的client將得到通知  

  82.             // 至關於釋放鎖  

  83.             zk.delete(this.thisPath, -1);  

  84.         }  

  85.     }  

  86.   

  87.     public static void main(String[] args) throws Exception {  

  88.         for (int i = 0; i < 10; i++) {  

  89.             new Thread() {  

  90.                 public void run() {  

  91.                     try {  

  92.                         DistributedClient dl = new DistributedClient();  

  93.                         dl.connectZookeeper();  

  94.                     } catch (Exception e) {  

  95.                         e.printStackTrace();  

  96.                     }  

  97.                 }  

  98.             }.start();  

  99.         }  

  100.   

  101.         Thread.sleep(Long.MAX_VALUE);  

  102.     }  

  103. }   

思考

思惟縝密的朋友可能會想到, 上述的方案並不安全. 假設某個client在得到鎖以前掛掉了, 因爲client建立的節點是ephemeral類型的, 所以這個節點也會被刪除, 從而致使排在這個client以後的client提早得到了鎖. 此時會存在多個client同時訪問共享資源.
如何解決這個問題呢? 能夠在接到waitPath的刪除通知的時候, 進行一次確認, 確認當前的thisPath是否真的是列表中最小的節點.

  1. // 發生了waitPath的刪除事件  

  2. if (event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {  

  3.     // 確認thisPath是否真的是列表中的最小節點  

  4.     List<String> childrenNodes = zk.getChildren("/" + groupNode, false);  

  5.     String thisNode = thisPath.substring(("/" + groupNode + "/").length());  

  6.     // 排序  

  7.     Collections.sort(childrenNodes);  

  8.     int index = childrenNodes.indexOf(thisNode);  

  9.     if (index == 0) {  

  10.         // 確實是最小節點  

  11.         doSomething();  

  12.     } else {  

  13.         // 說明waitPath是因爲出現異常而掛掉的  

  14.         // 更新waitPath  

  15.         waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);  

  16.         // 從新註冊監聽, 並判斷此時waitPath是否已刪除  

  17.         if (zk.exists(waitPath, true) == null) {  

  18.             doSomething();  

  19.         }  

  20.     }  

  21. }  

另外, 因爲thisPath和waitPath這2個成員變量會在多個線程中訪問, 最好將他們聲明爲volatile, 以防止出現線程可見性問題.

另外一種思路

下面介紹一種更簡單, 可是不怎麼推薦的解決方案.
每一個client在getChildren的時候, 註冊監聽子節點的變化. 當子節點的變化通知到來時, 再一次經過getChildren獲取子節點列表, 判斷thisPath是不是列表中的最小節點, 若是是, 則執行資源訪問邏輯.

  1. public class DistributedClient2 {  

  2.     // 超時時間  

  3.     private static final int SESSION_TIMEOUT = 5000;  

  4.     // zookeeper server列表  

  5.     private String hosts = "localhost:4180,localhost:4181,localhost:4182";  

  6.     private String groupNode = "locks";  

  7.     private String subNode = "sub";  

  8.   

  9.     private ZooKeeper zk;  

  10.     // 當前client建立的子節點  

  11.     private volatile String thisPath;  

  12.   

  13.     private CountDownLatch latch = new CountDownLatch(1);  

  14.   

  15.     /** 

  16.      * 鏈接zookeeper 

  17.      */  

  18.     public void connectZookeeper() throws Exception {  

  19.         zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {  

  20.             public void process(WatchedEvent event) {  

  21.                 try {  

  22.                     // 鏈接創建時, 打開latch, 喚醒wait在該latch上的線程  

  23.                     if (event.getState() == KeeperState.SyncConnected) {  

  24.                         latch.countDown();  

  25.                     }  

  26.   

  27.                     // 子節點發生變化  

  28.                     if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {  

  29.                         // thisPath是不是列表中的最小節點  

  30.                         List<String> childrenNodes = zk.getChildren("/" + groupNode, true);  

  31.                         String thisNode = thisPath.substring(("/" + groupNode + "/").length());  

  32.                         // 排序  

  33.                         Collections.sort(childrenNodes);  

  34.                         if (childrenNodes.indexOf(thisNode) == 0) {  

  35.                             doSomething();  

  36.                         }  

  37.                     }  

  38.                 } catch (Exception e) {  

  39.                     e.printStackTrace();  

  40.                 }  

  41.             }  

  42.         });  

  43.   

  44.         // 等待鏈接創建  

  45.         latch.await();  

  46.   

  47.         // 建立子節點  

  48.         thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,  

  49.                 CreateMode.EPHEMERAL_SEQUENTIAL);  

  50.   

  51.         // wait一小會, 讓結果更清晰一些  

  52.         Thread.sleep(10);  

  53.   

  54.         // 監聽子節點的變化  

  55.         List<String> childrenNodes = zk.getChildren("/" + groupNode, true);  

  56.   

  57.         // 列表中只有一個子節點, 那確定就是thisPath, 說明client得到鎖  

  58.         if (childrenNodes.size() == 1) {  

  59.             doSomething();  

  60.         }  

  61.     }  

  62.   

  63.     /** 

  64.      * 共享資源的訪問邏輯寫在這個方法中 

  65.      */  

  66.     private void doSomething() throws Exception {  

  67.         try {  

  68.             System.out.println("gain lock: " + thisPath);  

  69.             Thread.sleep(2000);  

  70.             // do something  

  71.         } finally {  

  72.             System.out.println("finished: " + thisPath);  

  73.             // 將thisPath刪除, 監聽thisPath的client將得到通知  

  74.             // 至關於釋放鎖  

  75.             zk.delete(this.thisPath, -1);  

  76.         }  

  77.     }  

  78.   

  79.     public static void main(String[] args) throws Exception {  

  80.         for (int i = 0; i < 10; i++) {  

  81.             new Thread() {  

  82.                 public void run() {  

  83.                     try {  

  84.                         DistributedClient2 dl = new DistributedClient2();  

  85.                         dl.connectZookeeper();  

  86.                     } catch (Exception e) {  

  87.                         e.printStackTrace();  

  88.                     }  

  89.                 }  

  90.             }.start();  

  91.         }  

  92.   

  93.         Thread.sleep(Long.MAX_VALUE);  

  94.     }  

  95. }  

爲何不推薦這個方案呢? 是由於每次子節點的增長和刪除都要廣播給全部client, client數量很少時還看不出問題. 若是存在不少client, 那麼就可能致使廣播風暴--過多的廣播通知阻塞了網絡. 使用第一個方案, 會使得通知的數量大大降低. 固然第一個方案更復雜一些, 複雜的方案同時也意味着更容易引進bug.

相關文章
相關標籤/搜索