分布式鎖2 Java非常用技術方案探討之ZooKeeper

前言:

      由於在平時的工作中，線上服務器是分布式多台部署的，經常會面臨解決分布式場景下數據一致性的問題，那么就要利用分布式鎖來解決這些問題。以自己結合實際工作中的一些經驗和網上看到的一些資料，做一個講解和總結。之前我已經寫了一篇關於分布式鎖的文章: 分布式鎖1 Java常用技術方案 。上一篇文章中主要寫的是在日常項目中，較為常見的幾種實現分布式鎖的方法。通過這些方法，基本上可以解決我們日常工作中大部分場景下使用分布式鎖的問題。

      本篇文章主要是在上一篇文章的基礎上，介紹一些雖然日常工作中不常用或者比較實現起來比較重，但是可以作為技術方案學習了解一下的分布式鎖方案。希望這篇文章可以方便自己以后查閱，同時要是能幫助到他人那也是很好的。

 

===============================================================長長的分割線====================================================================

 

正文:

      第一步，使用zookeeper節點名稱唯一性，用於分布式鎖:

      關於zookeeper集群的搭建，可以參考我之前寫的一篇文章: ZooKeeper1 利用虛擬機搭建自己的ZooKeeper集群

      zookeeper抽象出來的節點結構是一個和文件系統類似的小型的樹狀的目錄結構，同時zookeeper機制規定：同一個目錄下只能有一個唯一的文件名。例如：我們在zookeeper的根目錄下，由兩個客戶端同時創建一個名為/myDistributeLock，只有一個客戶端可以成功。

      上述方案和memcached的add()方法、redis的setnx()方法實現分布式鎖有着相同的思路。這樣的方案實現起來如果不考慮搭建和維護zookeeper集群的成本，由於正確性和可靠性是zookeeper機制自己保證的，實現還是比較簡單的。

      

 　  第二步，使用zookeeper臨時順序節點，用於分布式鎖:

 

      在討論這套方案之前，我們有必要先“吹毛求疵”般的說明一下使用zookeeper節點名稱唯一性來做分布式鎖這個方案的缺點。比如，當許多線程在等待一個鎖時，如果鎖得到釋放的時候，那么所有客戶端都被喚醒，但是僅僅有一個客戶端得到鎖。在這個過程中，大量的線程根本沒有獲得鎖的可能性，但是也會引起大量的上下文切換，這個系統開銷也是不小的，對於這樣的現象有一個專業名詞，稱之為“驚群效應”。

     我們首先說明一下zookeeper的順序節點、臨時節點和watcher機制:

     所謂順序節點，假如我們在/myDisLocks/目錄下創建3個節點，zookeeper集群會按照發起創建的順序來創建節點，節點分別為/myDisLocks/0000000001、/myDisLocks/0000000002、/myDisLocks/0000000003。

     所謂臨時節點，臨時節點由某個客戶端創建，當客戶端與zookeeper集群斷開連接，則該節點自動被刪除。

     所謂對於watcher機制，大家可以參考Apache ZooKeeper Watcher機制源碼解釋。當然如果你之前不知道watcher機制是個什么東東，不建議你直接去看前邊我提供的文章鏈接，這樣你極有可能忘掉我們的討論主線，即分布式鎖的實現方案，而陷入到watcher機制的源碼實現中。所以你也可以先看看下面的具體方案，猜測一下watcher是用來干嘛的，我這里先總結一句話做個引子: 所謂watcher機制，你可以簡單一點兒理解成任何一個連接zookeeper的客戶端可以通過watcher機制關注自己感興趣的節點的增刪改查，當這個節點發生增刪改查的操作時，會“廣播”自己的消息，所有對此感興趣的節點可以在收到這些消息后，根據自己的業務需要執行后續的操作。

     具體的使用步驟如下:

      1. 每個業務線程調用create()方法創建名為“/myDisLocks/thread”的節點，需要注意的是，這里節點的創建類型需要設置為EPHEMERAL_SEQUENTIAL，即節點類型為臨時順序節點。此時/myDisLocks節點下會出現諸如/myDisLocks/thread0000000001、/myDisLocks/thread0000000002、/myDisLocks/thread0000000003這樣的子節點。

     2. 每個業務線程調用getChildren(“myDisLocks”)方法來獲取/myDisLocks這個節點下所有已經創建的子節點。

      3. 每個業務線程獲取到所有子節點的路徑之后，如果發現自己在步驟1中創建的節點的尾綴編號是所有節點中序號最小的，那么就認為自己獲得了鎖。

      4. 如果在步驟3中發現自己並非是所有子節點中序號最小的，說明自己還沒有獲取到鎖。使用watcher機制監視比自己創建節點的序列號小的節點（比自己創建的節點小的最大節點），進入等待。比如，如果當前業務線程創建的節點是/myDisLocks/thread0000000003，那么在沒有獲取到鎖的情況下，他只需要監視/myDisLocks/thread0000000002的情況。只有當/myDisLocks/thread0000000002獲取到鎖並釋放之后，當前業務線程才啟動獲取鎖，這樣可以避免一個業務線程釋放鎖之后，其他所有線程都去競爭鎖，引起不必要的上下文切換，最終造成“驚群現象”。

     5. 釋放鎖的過程相對比較簡單，就是刪除自己創建的那個子節點即可。

      注意: 這個方案實現的分布式鎖還帶着一點兒公平鎖的味道！為什么呢？我們在利用每個節點的序號進行排隊以此來避免進群現象時，實際上所有業務線程獲得鎖的順序就是自己創建節點的順序，也就是哪個業務線程先來，哪個就可以最快獲得鎖。

      下面貼出我自己實現的上述方案的代碼:

      1. 代碼中有兩個Java類: MyDistributedLockByZK.java和LockWatcher.java。其中MyDistributedLockByZK.java中的main函數利用線程池啟動5個線程，以此來模擬多個業務線程競爭鎖的情況；而LockWatcher.java定義分布式鎖和實現了watcher機制。

      2. 同時，我使用的zookeeper集群是自己以前利用VMWare搭建的集群，所以zookeeper鏈接是192.168.224.170:2181，大家可以根據替換成自己的zookeeper鏈接即可。

public class MyDistributedLockByZK {
    /** 線程池 **/
    private static ExecutorService executorService = null;
    private static final int THREAD_NUM = 5;
    private static int threadNo = 0;
    private static CountDownLatch threadCompleteLatch = new CountDownLatch(THREAD_NUM);
    
    /** ZK的相關配置常量 **/
    private static final String CONNECTION_STRING = "192.168.224.170:2181";
    private static final int SESSION_TIMEOUT = 10000;
    // 此變量在LockWatcher中也有一個同名的靜態變量，正式使用的時候，提取到常量類中共同維護即可。
    private static final String LOCK_ROOT_PATH = "/myDisLocks";
    
    public static void main(String[] args) {
        // 定義線程池
        executorService = Executors.newFixedThreadPool(THREAD_NUM, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                String name = String.format("第[%s]個測試線程", ++threadNo);
                Thread ret = new Thread(Thread.currentThread().getThreadGroup(), r, name, 0);
                ret.setDaemon(false);
                return ret;
            }
        });
        
        // 啟動線程
        if (executorService != null) {
            startProcess();
        }
    }
    
    /**
     * @author zhangyi03
     * @date 2017-5-23 下午5:57:27
     * @description 模擬並發執行任務
     */
     public static void startProcess() {    
        Runnable disposeBusinessRunnable= new Thread(new Runnable() {
            public void run() {
                String threadName = Thread.currentThread().getName();
                
                LockWatcher lock = new LockWatcher(threadCompleteLatch);
                try {
                    /** 步驟1: 當前線程創建ZK連接  **/
                    lock.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
                    
                    /** 步驟2: 創建鎖的根節點  **/
                    // 注意，此處創建根節點的方式其實完全可以在初始化的時候由主線程單獨進行根節點的創建，沒有必要在業務線程中創建。
                    // 這里這樣寫只是一種思路而已，不必局限於此
                    synchronized (MyDistributedLockByZK.class){
                        lock.createPersistentPath(LOCK_ROOT_PATH, "該節點由" + threadName + "創建", true);
                    }
                    
                    /** 步驟3: 開啟鎖競爭並執行任務 **/
                    lock.getLock();
                } catch (Exception e) {
                    e.printStackTrace();
                } 
            }  
        });
        
        for (int i = 0; i < THREAD_NUM; i++) {
            executorService.execute(disposeBusinessRunnable);
        }
        executorService.shutdown();
        
        try {
            threadCompleteLatch.await();
            System.out.println("所有線程運行結束!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
     }
}

public class LockWatcher implements Watcher {
    /** 成員變量 **/
    private ZooKeeper zk = null;
    // 當前業務線程競爭鎖的時候創建的節點路徑
    private String selfPath = null;
    // 當前業務線程競爭鎖的時候創建節點的前置節點路徑
    private String waitPath = null;
    // 確保連接zk成功；只有當收到Watcher的監聽事件之后，才執行后續的操作，否則請求阻塞在createConnection()創建ZK連接的方法中
    private CountDownLatch connectSuccessLatch = new CountDownLatch(1);
    // 標識線程是否執行完任務
    private CountDownLatch threadCompleteLatch = null;
    
    /** ZK的相關配置常量 **/
    private static final String LOCK_ROOT_PATH = "/myDisLocks";
    private static final String LOCK_SUB_PATH = LOCK_ROOT_PATH + "/thread";
    
    public LockWatcher(CountDownLatch latch) {
        this.threadCompleteLatch = latch;
    }
    
    @Override
    public void process(WatchedEvent event) {
        if (event == null) {
            return;
        }
        
        // 通知狀態
        Event.KeeperState keeperState = event.getState();
        // 事件類型
        Event.EventType eventType = event.getType();
        
        // 根據通知狀態分別處理
        if (Event.KeeperState.SyncConnected == keeperState) {
            if ( Event.EventType.None == eventType ) {
                System.out.println(Thread.currentThread().getName() + "成功連接上ZK服務器");
                // 此處代碼的主要作用是用來輔助判斷當前線程確實已經連接上ZK
                connectSuccessLatch.countDown();
            }else if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                System.out.println(Thread.currentThread().getName() + "收到情報，排我前面的家伙已掛，我准備再次確認我是不是最小的節點！？");
                try {
                    if(checkMinPath()){
                        getLockSuccess();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } 
            }
        } else if ( Event.KeeperState.Disconnected == keeperState ) {
            System.out.println(Thread.currentThread().getName() + "與ZK服務器斷開連接");
        } else if ( Event.KeeperState.AuthFailed == keeperState ) {
            System.out.println(Thread.currentThread().getName() + "權限檢查失敗");
        } else if ( Event.KeeperState.Expired == keeperState ) {
            System.out.println(Thread.currentThread().getName() + "會話失效");
        }
    }
    
     /**
     * @author zhangyi03
     * @date 2017-5-23 下午6:07:03
     * @description 創建ZK連接
     * @param connectString ZK服務器地址列表
     * @param sessionTimeout Session超時時間
     * @throws IOException
     * @throws InterruptedException
     */
    public void createConnection(String connectString, int sessionTimeout) throws IOException, InterruptedException {
        zk = new ZooKeeper(connectString, sessionTimeout, this);
        // connectSuccessLatch.await(1, TimeUnit.SECONDS) 正式實現的時候可以考慮此處是否采用超時阻塞
        connectSuccessLatch.await();
    }
    
    /**
     * @author zhangyi03
     * @date 2017-5-23 下午6:15:48
     * @description 創建ZK節點
     * @param path 節點path
     * @param data 初始數據內容
     * @param needWatch
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    public boolean createPersistentPath(String path, String data, boolean needWatch) throws KeeperException, InterruptedException {
        if(zk.exists(path, needWatch) == null){
            String result = zk.create( path,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println(Thread.currentThread().getName() + "創建節點成功, path: " + result + ", content: " + data);
        }
        return true;
    }
    
    /**
     * @author zhangyi03
     * @date 2017-5-23 下午6:24:46
     * @description 獲取分布式鎖
     * @throws KeeperException
     * @throws InterruptedException
     */
     public void getLock() throws Exception {
        selfPath = zk.create(LOCK_SUB_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(Thread.currentThread().getName() + "創建鎖路徑:" + selfPath);
        if(checkMinPath()){
            getLockSuccess();
        }
     }
     
     /**
     * @author zhangyi03
     * @date 2017-5-23 下午7:02:41
     * @description 獲取鎖成功
     * @throws KeeperException
     * @throws InterruptedException
     */
    private void getLockSuccess() throws KeeperException, InterruptedException {
         if(zk.exists(selfPath, false) == null){
             System.err.println(Thread.currentThread().getName() + "本節點已不在了...");
             return;
         }
         System.out.println(Thread.currentThread().getName() + "獲取鎖成功，開始處理業務數據！");
         Thread.sleep(2000);
         System.out.println(Thread.currentThread().getName() + "處理業務數據完成，刪除本節點：" + selfPath);
         zk.delete(selfPath, -1);
         releaseConnection();
         threadCompleteLatch.countDown();
     }

     /**
     * @author zhangyi03
     * @date 2017-5-23 下午7:06:46
     * @description 關閉ZK連接
     */
    private void releaseConnection() {
        if (zk != null) {
            try {
                zk.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println(Thread.currentThread().getName() + "釋放ZK連接");
     }

     /**
     * @author zhangyi03
     * @date 2017-5-23 下午6:57:14
     * @description 檢查自己是不是最小的節點
     * @param selfPath
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    private boolean checkMinPath() throws Exception {
          List<String> subNodes = zk.getChildren(LOCK_ROOT_PATH, false);
          // 根據元素按字典序升序排序
          Collections.sort(subNodes);
          System.err.println(Thread.currentThread().getName() + "創建的臨時節點名稱:" + selfPath.substring(LOCK_ROOT_PATH.length()+1));
          int index = subNodes.indexOf(selfPath.substring(LOCK_ROOT_PATH.length()+1));
          System.err.println(Thread.currentThread().getName() + "創建的臨時節點的index:" + index);
          switch (index){
              case -1: {
                  System.err.println(Thread.currentThread().getName() + "創建的節點已不在了..." + selfPath);
                  return false;
              }
              case 0:{
                  System.out.println(Thread.currentThread().getName() +  "子節點中，我果然是老大" + selfPath);
                  return true;
              }
              default:{
                  // 獲取比當前節點小的前置節點,此處只關注前置節點是否還在存在，避免驚群現象產生
                  waitPath = LOCK_ROOT_PATH +"/"+ subNodes.get(index - 1);
                  System.out.println(Thread.currentThread().getName() + "獲取子節點中，排在我前面的節點是:" + waitPath);
                  try {
                      zk.getData(waitPath, true, new Stat());
                      return false;
                  } catch (Exception e) {
                      if (zk.exists(waitPath, false) == null) {
                          System.out.println(Thread.currentThread().getName() + "子節點中，排在我前面的" + waitPath + "已失蹤，該我了");
                          return checkMinPath();
                      } else {
                          throw e;
                      }
                  }
              }
                  
          }
     }
}

 

       第三步，使用memcached的cas()方法，用於分布式鎖:

       下篇文章我們再細說！

 

       第四步，使用redis的watch、multi、exec命令，用於分布式鎖:

       下篇文章我們再細說！

 

       第五步，總結:

      綜上，對於分布式鎖這些非常用或者實現起來比較重的方案，大家可以根據自己在項目中的需要，酌情使用。最近在和別人討論的過程中，以及我的第一篇關於分布式鎖的文章分布式鎖1 Java常用技術方案  大家的回復中，總結來看，對於用redis實現分布式鎖確實存在着比較多的細節問題可以進行深入討論，歡迎大家留言，相互學習。

      忍不住嘚瑟一下，我媳婦兒此刻在我旁邊看AbstractQueuedSynchronizer，厲害吧？！，一會兒出去吃飯，哈哈~

 

      第六步，線上使用補充篇:

       截止到2017.08.25(周五)，使用上述文章中的”臨時節點+watcher機制方案”解決一個分布式鎖的問題時，最終發現在實現過程中，由於watcher機制類似於通知等待機制的特點，如果主線程在經歷“獲取鎖操作”、“處理業務代碼”、“釋放鎖操作”這三步的過程中，使用watcher機制阻塞的獲取鎖時，會導致根本無法將獲取鎖結果返回給主線程，而在實際的時候過程中，一般情況下主線程在“獲取鎖操作”時都希望可以同步獲得一個返回值。

      所以，上述的”臨時節點+watcher機制方案”從技術方案角度足夠完美，但是在實際使用過程中，個人覺得還不是特別的方便。

      

      轉載請注明來自博客園http://www.cnblogs.com/PurpleDream/p/5573040.html ，版權歸本人和博客園所有，謝謝！