分布式任務&分布式鎖（li），分布式任務調度框架

分布式任務&分布式鎖（li），分布式任務調度框架

　　目前系統中存在批量審批、批量授權等各個操作，批量操作中可能因為處理機器、線程不同，造成刷新緩存丟失授權等信息，如批量審批同一用戶權限多個權限申請後，流程平台並發的發送多個http請求到acl不同服務器，a機器處理了授權a，b機器同時處理了授權b，然後刷新用戶緩存。因為在事務裡彼此看不見對方提交的數據，刷新時又完全從db中讀取要刷新的數據，就造成了互相丟失對方的數據。因此，需要一個分布式鎖工具，來協調各個機器、線上的工作同步問題。

　　分布式鎖千萬不能用ReentrantLock，因為它的lock和unlock要求在同一線程中處理。而我們的鎖是分布式的，不一定在同一個機器上執行。

　　Zookeeper作為一個分布式的服務框架，主要用來解決分布式集群中應用系統的一致性問題，它能提供基於類似於文件系統的目錄節點樹方式的數據存儲，但是 Zookeeper並不是用來專門存儲數據的，它的作用主要是用來維護和監控你存儲的數據的狀態變化。通過監控這些數據狀態的變化，從而可以達到基於數據的集群管理，後面將會詳細介紹 Zookeeper 能夠解決的一些典型問題，這裡先介紹一下，Zookeeper 的操作接口和簡單使用示例。

一、常用接口列表

　　客戶端要連接Zookeeper服務器可以通過創建org.apache.zookeeper.ZooKeeper的一個實例對象，然後調用這個類提供的接口來和服務器交互。前面說了ZooKeeper主要是用來維護和監控一個目錄節點樹中存儲的數據的狀態，所有我們能夠操作 ZooKeeper 的也和操作目錄節點樹大體一樣，如創建一個目錄節點，給某個目錄節點設置數據，獲取某個目錄節點的所有子目錄節點，給某個目錄節點設置權限和監控這個目錄節點的狀態變化。

　　org.apache.zookeeper.ZooKeeper方法列表如下，方法名及方法功能描述：

 除以上這些上表中列出的方法之外還有一些重載方法，如都提供了一個回調類的重載方法以及可以設置特定 Watcher 的重載方法，具體的方法可以參考 org.apache.zookeeper.ZooKeeper 類的 API 說明。

二、基本操作

　　下面給出基本的操作ZooKeeper的示例代碼，這樣你就能對ZooKeeper有直觀的認識了。下面的清單包括了創建與 ZooKeeper 服務器的連接以及最基本的數據操作：

//創建一個與服務器的連接
ZooKeeper zk = new ZooKeeper("localhost:" + CLIENT_PORT, 
     ClientBase.CONNECTION_TIMEOUT, new Watcher() { 
         // 監控所有被觸發的事件
         public void process(WatchedEvent event) { 
             System.out.println("已經觸發了" + event.getType() + "事件！"); 
         } 
     }); 
// 創建一個目錄節點
zk.create("/testRootPath", "testRootData".getBytes(), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT); 
// 創建一個子目錄節點
zk.create("/testRootPath/testChildPathOne", "testChildDataOne".getBytes(),
Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); 
System.out.println(new String(zk.getData("/testRootPath",false,null))); 
// 取出子目錄節點列表
System.out.println(zk.getChildren("/testRootPath",true)); 
// 修改子目錄節點數據
zk.setData("/testRootPath/testChildPathOne","modifyChildDataOne".getBytes(),-1); 
System.out.println("目錄節點狀態：["+zk.exists("/testRootPath",true)+"]"); 
// 創建另外一個子目錄節點
zk.create("/testRootPath/testChildPathTwo", "testChildDataTwo".getBytes(), 
Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); 
System.out.println(new String(zk.getData("/testRootPath/testChildPathTwo",true,null))); 
// 刪除子目錄節點
zk.delete("/testRootPath/testChildPathTwo",-1); 
zk.delete("/testRootPath/testChildPathOne",-1); 
// 刪除父目錄節點
zk.delete("/testRootPath",-1); 
// 關閉連接
zk.close(); 

// #輸出的結果如下：
// 已經觸發了 None 事件！
// testRootData
// [testChildPathOne]
// 目錄節點狀態：[5,5,1281804532336,1281804532336,0,1,0,0,12,1,6]
// 已經觸發了 NodeChildrenChanged 事件！
// testChildDataTwo
// 已經觸發了 NodeDeleted 事件！
// 已經觸發了 NodeDeleted 事件！

當對目錄節點監控狀態打開時，一旦目錄節點的狀態發生變化，Watcher 對象的 process 方法就會被調用。

三、Zookeeper的應用場景

　　Zookeeper從設計模式角度來看，是一個基於觀察者模式設計的分布式服務管理框架，它負責存儲和管理大家都關心的數據，然後接受觀察者的注冊，一旦這些數據的狀態發生變化，Zookeeper就將負責通知已經在 Zookeeper上注冊的那些觀察者做出相應的反應，從而實現集群中類似 Master/Slave 管理模式，關於 Zookeeper的詳細架構等內部細節可以閱讀 Zookeeper的源碼。下面詳細介紹這些典型的應用場景，也就是 Zookeeper 到底能幫我們解決那些問題？下面將給出答案。

　　統一命名服務（Name Service），分布式應用中，通常需要有一套完整的命名規則，既能夠產生唯一的名稱又便於人識別和記住，通常情況下用樹形的名稱結構是一個理想的選擇，樹形的名稱結構是一個有層次的目錄結構，既對人友好又不會重復。說到這裡你可能想到了 JNDI，沒錯 Zookeeper 的 Name Service 與 JNDI 能夠完成的功能是差不多的，它們都是將有層次的目錄結構關聯到一定資源上，但是 Zookeeper 的 Name Service 更加是廣泛意義上的關聯，也許你並不需要將名稱關聯到特定資源上，你可能只需要一個不會重復名稱，就像數據庫中產生一個唯一的數字主鍵一樣。Name Service已經是 Zookeeper 內置的功能，你只要調用 Zookeeper 的 API 就能實現。如調用 create 接口就可以很容易創建一個目錄節點。

　　配置管理，配置的管理在分布式應用環境中很常見，例如同一個應用系統需要多台 PC Server 運行，但是它們運行的應用系統的某些配置項是相同的，如果要修改這些相同的配置項，那麼就必須同時修改每台運行這個應用系統的 PC Server，這樣非常麻煩而且容易出錯。像這樣的配置信息完全可以交給 Zookeeper 來管理，將配置信息保存在 Zookeeper 的某個目錄節點中，然後將所有需要修改的應用機器監控配置信息的狀態，一旦配置信息發生變化，每台應用機器就會收到 Zookeeper 的通知，然後從 Zookeeper 獲取新的配置信息應用到系統中。配置管理結構圖如下：

　　集群管理，Zookeeper 能夠很容易的實現集群管理的功能，如有多台 Server 組成一個服務集群，那麼必須要一個“總管”知道當前集群中每台機器的服務狀態，一旦有機器不能提供服務，集群中其它集群必須知道，從而做出調整重新分配服務策略。同樣當增加集群的服務能力時，就會增加一台或多台 Server，同樣也必須讓“總管”知道。Zookeeper 不僅能夠幫你維護當前的集群中機器的服務狀態，而且能夠幫你選出一個“總管”，讓這個總管來管理集群，這就是 Zookeeper 的另一個功能 Leader Election。它們的實現方式都是在 Zookeeper 上創建一個 EPHEMERAL 類型的目錄節點，然後每個 Server 在它們創建目錄節點的父目錄節點上調用 getChildren(String path, boolean watch) 方法並設置 watch 為 true，由於是 EPHEMERAL 目錄節點，當創建它的 Server 死去，這個目錄節點也隨之被刪除，所以 Children 將會變化，這時 getChildren上的 Watch 將會被調用，所以其它 Server 就知道已經有某台 Server 死去了。新增 Server 也是同樣的原理。

　　Zookeeper如何實現 Leader Election，也就是選出一個 Master Server。和前面的一樣每台 Server 創建一個 EPHEMERAL 目錄節點，不同的是它還是一個 SEQUENTIAL 目錄節點，所以它是個 EPHEMERAL_SEQUENTIAL 目錄節點。之所以它是 EPHEMERAL_SEQUENTIAL 目錄節點，是因為我們可以給每台 Server 編號，我們可以選擇當前是最小編號的 Server 為 Master，假如這個最小編號的 Server 死去，由於是 EPHEMERAL 節點，死去的 Server 對應的節點也被刪除，所以當前的節點列表中又出現一個最小編號的節點，我們就選擇這個節點為當前 Master。這樣就實現了動態選擇 Master，避免了傳統意義上單 Master 容易出現單點故障的問題。下圖是集群管理結構圖：

　　這部分的示例代碼如下：

 void findLeader() throws InterruptedException { 
        byte[] leader = null; 
        try { 
            leader = zk.getData(root + "/leader", true, null); 
        } catch (Exception e) { 
            logger.error(e); 
        } 
        if (leader != null) { 
            following(); 
        } else { 
            String newLeader = null; 
            try { 
                byte[] localhost = InetAddress.getLocalHost().getAddress(); 
                newLeader = zk.create(root + "/leader", localhost, 
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); 
            } catch (Exception e) { 
                logger.error(e); 
            } 
            if (newLeader != null) { 
                leading(); 
            } else { 
                mutex.wait(); 
            } 
        } 
    } 

 void getLock() throws KeeperException, InterruptedException{ 
        List<String> list = zk.getChildren(root, false); 
        String[] nodes = list.toArray(new String[list.size()]); 
        Arrays.sort(nodes); 
        if(myZnode.equals(root+"/"+nodes[0])){ 
            doAction(); 
        } 
        else{ 
            waitForLock(nodes[0]); 
        } 
    } 
    void waitForLock(String lower) throws InterruptedException, KeeperException {
        Stat stat = zk.exists(root + "/" + lower,true); 
        if(stat != null){ 
            mutex.wait(); 
        } 
        else{ 
            getLock(); 
        } 
    }

//#加鎖：
//ZooKeeper 將按照如下方式實現加鎖的操作：
//1 ） ZooKeeper 調用 create （）方法來創建一個路徑格式為“ _locknode_/lock- ”的節點，此節點類型為sequence （連續）和 ephemeral （臨時）。
　　　　也就是說，創建的節點為臨時節點，並且所有的節點連續編號，即“ lock-i ”的格式。
//2 ）在創建的鎖節點上調用 getChildren （）方法，來獲取鎖目錄下的最小編號節點，並且不設置 watch 。
//3 ）步驟 2 中獲取的節點恰好是步驟 1 中客戶端創建的節點，那麼此客戶端獲得此種類型的鎖，然後退出操作。
//4 ）客戶端在鎖目錄上調用 exists （）方法，並且設置 watch 來監視鎖目錄下比自己小一個的連續臨時節點的狀態。
//5 ）如果監視節點狀態發生變化，則跳轉到第 2 步，繼續進行後續的操作，直到退出鎖競爭。
// 
//#解鎖： 
//ZooKeeper 解鎖操作非常簡單，客戶端只需要將加鎖操作步驟 1 中創建的臨時節點刪除即可。

 void addQueue() throws KeeperException, InterruptedException{ 
        zk.exists(root + "/start",true); 
        zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE, 
        CreateMode.EPHEMERAL_SEQUENTIAL); 
        synchronized (mutex) { 
            List<String> list = zk.getChildren(root, false); 
            if (list.size() < size) { 
                mutex.wait(); 
            } else { 
                zk.create(root + "/start", new byte[0], Ids.OPEN_ACL_UNSAFE,
                 CreateMode.PERSISTENT); 
            } 
        } 
 }

 public void process(WatchedEvent event) { 
        if(event.getPath().equals(root + "/start") &&
         event.getType() == Event.EventType.NodeCreated){ 
            System.out.println("得到通知"); 
            super.process(event); 
            doAction(); 
        } 
    } 

#生產者代碼
boolean produce(int i) throws KeeperException, InterruptedException{ 
        ByteBuffer b = ByteBuffer.allocate(4); 
        byte[] value; 
        b.putInt(i); 
        value = b.array(); 
        zk.create(root + "/element", value, ZooDefs.Ids.OPEN_ACL_UNSAFE, 
                    CreateMode.PERSISTENT_SEQUENTIAL); 
        return true; 
    } 

 
#消費者代碼
int consume() throws KeeperException, InterruptedException{ 
        int retvalue = -1; 
        Stat stat = null; 
        while (true) { 
            synchronized (mutex) { 
                List<String> list = zk.getChildren(root, true); 
                if (list.size() == 0) { 
                    mutex.wait(); 
                } else { 
                    Integer min = new Integer(list.get(0).substring(7)); 
                    for(String s : list){ 
                        Integer tempValue = new Integer(s.substring(7)); 
                        if(tempValue < min) min = tempValue; 
                    } 
                    byte[] b = zk.getData(root + "/element" + min,false, stat); 
                    zk.delete(root + "/element" + min, 0); 
                    ByteBuffer buffer = ByteBuffer.wrap(b); 
                    retvalue = buffer.getInt(); 
                    return retvalue; 
                } 
            } 
        } 
 }