Curator簡介

Apache Curator是一個比較完善的ZooKeeper客戶端框架，通過封裝的一套高級API 簡化了ZooKeeper的操作。通過查看官方文檔，可以發(fā)現(xiàn)Curator主要解決了三類問題：

• 封裝ZooKeeper client與ZooKeeper server之間的連接處理
• 提供了一套Fluent風(fēng)格的操作API
• 提供ZooKeeper各種應(yīng)用場景(recipe， 比如：分布式鎖服務(wù)、集群領(lǐng)導(dǎo)選舉、共享計數(shù)器、緩存機制、分布式隊列等)的抽象封裝

Curator主要從以下幾個方面降低了zk使用的復(fù)雜性：

• 重試機制:提供可插拔的重試機制, 它將給捕獲所有可恢復(fù)的異常配置一個重試策略，并且內(nèi)部也提供了幾種標(biāo)準(zhǔn)的重試策略(比如指數(shù)補償)
• 連接狀態(tài)監(jiān)控: Curator初始化之后會一直對zk連接進(jìn)行監(jiān)聽，一旦發(fā)現(xiàn)連接狀態(tài)發(fā)生變化將會作出相應(yīng)的處理
• zk客戶端實例管理:Curator會對zk客戶端到server集群的連接進(jìn)行管理，并在需要的時候重建zk實例，保證與zk集群連接的可靠性
• 各種使用場景支持:Curator實現(xiàn)了zk支持的大部分使用場景（甚至包括zk自身不支持的場景），這些實現(xiàn)都遵循了zk的最佳實踐，并考慮了各種極端情況

采用Curator去是使用zookeeper往往操作簡單，代碼量少，可以很簡單地實現(xiàn)斷線重連，監(jiān)聽器等使用起來也比較簡單，還支持分布式鎖等一些zookeeper常用的功能的代碼實現(xiàn)。

使用Curator監(jiān)聽目錄的數(shù)據(jù)變化

集群中的機器去監(jiān)聽zookeeper某一個目錄的數(shù)據(jù)變化，就可以動態(tài)地、實時地將新的配置信息修改至集群的每一臺機器，省去了手工配置地麻煩，還可以在程序運行過程中動態(tài)地更新一些配置。采用Curator的NodeCache可以完成一次注冊n次監(jiān)聽，這個對象可以緩存節(jié)點數(shù)據(jù)，在節(jié)點數(shù)據(jù)發(fā)生改變的時候，就會觸發(fā)這個事件。

下面分別是修改配置的客戶端和監(jiān)聽器程序代碼：
修改的客戶端代碼：

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;

public class myTest {
    /** zookeeper地址 */
    static final String CONNECT_ADDR = "localhost:2181";
    /** session超時時間 */
    static final int SESSION_OUTTIME = 5000;//ms 
    public static void main(String[] args) throws Exception {
        String path = "/myApplication/myIPs";
        //1 重試策略：初試時間為1s 重試10次
                RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
                //2 通過工廠創(chuàng)建連接
                CuratorFramework cf = CuratorFrameworkFactory.builder()
                            .connectString(CONNECT_ADDR)
                            .sessionTimeoutMs(SESSION_OUTTIME)
                            .retryPolicy(retryPolicy)
                            .build();
                //3 開啟連接
                cf.start();
                
                Stat stat = cf.checkExists().forPath(path);
                 
                if (stat == null)
                {
                    System.out.println("節(jié)點尚不存在");
                    cf.create().forPath(path, "192.168.1.102".getBytes());
                }
                cf.setData().forPath(path, "192.168.1.102,192.168.2.11,192.168.2.11".getBytes());
                cf.delete().forPath(path);
                if (cf != null) {
                     cf.close();
                    }
                
    }

}

監(jiān)聽器代碼：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

public class CuratorWatcher1 {
    
    private static List<String> ips = new ArrayList<String>();
    
    /** zookeeper地址 */
    static final String CONNECT_ADDR = "localhost:2181";
    /** session超時時間 */
    static final int SESSION_OUTTIME = 5000;//ms 
    
    public static void main(String[] args) throws Exception {
        
        //1 重試策略：初試時間為1s 重試10次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
        //2 通過工廠創(chuàng)建連接
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                    .connectString(CONNECT_ADDR)
                    .sessionTimeoutMs(SESSION_OUTTIME)
                    .retryPolicy(retryPolicy)
                    .build();
        
        //3 建立連接
        cf.start();
        /**
         * 想要實現(xiàn)watch一次注冊n次監(jiān)聽的話，我們需要使用到curator里的一個NodeCache對象。
         * 這個對象可以用來緩存節(jié)點數(shù)據(jù)，并且可以給節(jié)點添加nodeChange事件，當(dāng)節(jié)點的數(shù)據(jù)發(fā)生變化就會觸發(fā)這個事件
         */
        //4 建立一個cache緩存    Curator之nodeCache一次注冊，N次監(jiān)聽
        final NodeCache cache = new NodeCache(cf, "/myApplication/myIPs", false);
        cache.start(true);
        cache.getListenable().addListener(new NodeCacheListener() {
            /**
             * <B>方法名稱：</B>nodeChanged<BR>
        
             * @see org.apache.curator.framework.recipes.cache.NodeCacheListener#nodeChanged()
             */
            @Override
            public void nodeChanged() throws Exception {
                 // 防止節(jié)點被刪除時發(fā)生錯誤
                if (cache.getCurrentData() == null) {
                    System.out.println("獲取節(jié)點數(shù)據(jù)異常，無法獲取當(dāng)前緩存的節(jié)點數(shù)據(jù)，可能該節(jié)點已被刪除");
                    return;
                }

                 // 獲取節(jié)點最新的數(shù)據(jù)
                String data = new String(cache.getCurrentData().getData());
                System.out.println(cache.getCurrentData().getPath() + " 節(jié)點的數(shù)據(jù)發(fā)生變化，最新的數(shù)據(jù)為：" + data);
                if(data!=null){
                    String[] ipsArray = data.split(",");
                    System.out.println("change the Parameter ips !!!");
                    synchronized(ips){
                        ips.clear();
                        if(ipsArray!=null&&ipsArray.length>0){
                            for(int i=0;i<ipsArray.length;i++){
                                ips.add(ipsArray[i]);
                            }
                        }
                    }
                }
            
            }
        });
        Thread.sleep(Integer.MAX_VALUE);
        // 獲取當(dāng)前客戶端的狀態(tài)
        boolean isZkCuratorStarted = cf.isStarted();
        System.out.println("當(dāng)前客戶端的狀態(tài)：" + (isZkCuratorStarted ? "連接中..." : "已關(guān)閉..."));
    }
}

運行結(jié)果：

curator監(jiān)聽器.png

程序持續(xù)運行，就可以將全局的靜態(tài)參數(shù)ips 動態(tài)修改。可以看到，節(jié)點創(chuàng)建并賦值的過程，更新數(shù)據(jù)的時候以及刪除的時候都觸發(fā)了監(jiān)聽器。

當(dāng)然，這只是一個簡單的demo,很多實際的場景中，我們還得有參數(shù)修改后的重新加載過程，這里要根據(jù)需求來進(jìn)行編碼。