Zookeeper如何實現(xiàn)分布式鎖

排他鎖

排他鎖（Exclusive Locks），又被稱為寫鎖或獨占鎖，如果事務T1對數(shù)據對象O1加上排他鎖，那么整個加鎖期間，只允許事務T1對O1進行讀取和更新操作，其他任何事務都不能進行讀或寫。

定義鎖：

/exclusive_lock/lock

實現(xiàn)方式：

利用 zookeeper 的同級節(jié)點的唯一性特性，在需要獲取排他鎖時，所有的客戶端試圖通過調用 create() 接口，在 /exclusive_lock 節(jié)點下創(chuàng)建臨時子節(jié)點 /exclusive_lock/lock，最終只有一個客戶端能創(chuàng)建成功，那么此客戶端就獲得了分布式鎖。同時，所有沒有獲取到鎖的客戶端可以在 /exclusive_lock 節(jié)點上注冊一個子節(jié)點變更的 watcher 監(jiān)聽事件，以便重新爭取獲得鎖。

共享鎖

共享鎖（Shared Locks），又稱讀鎖。如果事務T1對數(shù)據對象O1加上了共享鎖，那么當前事務只能對O1進行讀取操作，其他事務也只能對這個數(shù)據對象加共享鎖，直到該數(shù)據對象上的所有共享鎖都釋放。

定義鎖:

/shared_lock/[hostname]-請求類型W/R-序號

實現(xiàn)方式： 1、客戶端調用 create 方法創(chuàng)建類似定義鎖方式的臨時順序節(jié)點。

2、客戶端調用 getChildren 接口來獲取所有已創(chuàng)建的子節(jié)點列表。

3、判斷是否獲得鎖，對于讀請求如果所有比自己小的子節(jié)點都是讀請求或者沒有比自己序號小的子節(jié)點，表明已經成功獲取共享鎖，同時開始執(zhí)行度邏輯。對于寫請求，如果自己不是序號最小的子節(jié)點，那么就進入等待。

4、如果沒有獲取到共享鎖，讀請求向比自己序號小的最后一個寫請求節(jié)點注冊 watcher 監(jiān)聽，寫請求向比自己序號小的最后一個節(jié)點注冊watcher 監(jiān)聽。

實際開發(fā)過程中，可以 curator 工具包封裝的API幫助我們實現(xiàn)分布式鎖。

 org.apache.curator
 curator-recipes
 x.x.x

curator 的幾種鎖方案 ：

1. InterProcessMutex：分布式可重入排它鎖
2. InterProcessSemaphoreMutex：分布式排它鎖
3. InterProcessReadWriteLock：分布式讀寫鎖

下面例子模擬 50 個線程使用重入排它鎖 InterProcessMutex 同時爭搶鎖：

實例

public class InterprocessLock {
   public static void main(String[] args)  {
       CuratorFramework zkClient = getZkClient();
       String lockPath = "/lock";
       InterProcessMutex lock = new InterProcessMutex(zkClient, lockPath);
       //模擬50個線程搶鎖
       for (int i = 0; i run() {
           try {
               lock.acquire();
               System.out.println("第"+threadFlag+"線程獲取到了鎖");
               //等到1秒后釋放鎖
               Thread.sleep(1000);
           } catch (Exception e) {
               e.printStackTrace();
           }finally {
               try {
                   lock.release();
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       }
   }

   private static CuratorFramework getZkClient() {
       String zkServerAddress = "192.168.3.39:2181";
       ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
       CuratorFramework zkClient = CuratorFrameworkFactory.builder()
               .connectString(zkServerAddress)
               .sessionTimeoutMs(5000)
               .connectionTimeoutMs(5000)
               .retryPolicy(retryPolicy)
               .build();
       zkClient.start();
       return zkClient;
   }
}

控制臺每間隔一秒鐘輸出一條記錄：