Curator入門教程1

 簡介

Curator是Netflix開源的一套ZooKeeper客戶端框架. Netflix在使用ZooKeeper的過程中發現ZooKeeper自帶的客戶端太底層, 應用方在使用的時候需要自己處理很多事情, 於是在它的基礎上包裝了一下, 提供了一套更好用的客戶端框架。

1、zookekeeper基本功能

     連接zookeeper,對節點進行crud、回調函數、判斷節點是否存在

package bjsxt.curator.base;

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.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;

public class CuratorBase {
    
    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.0.4:2181,192.168.0.9:2181,192.168.0.6: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 通過工廠創建連接
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                    .connectString(CONNECT_ADDR)
                    .sessionTimeoutMs(SESSION_OUTTIME)
                    .retryPolicy(retryPolicy)
//                    .namespace("super")
                    .build();
        //3 開啟連接
        cf.start();
        
        System.out.println(States.CONNECTED);
        System.out.println(cf.getState());
        
        // 新加、刪除
    
        //4 建立節點 指定節點類型（不加withMode默認為持久類型節點）、路徑、數據內容
        //cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c1","c1內容".getBytes());
        //5 刪除節點
         // cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");
        
        
        // 讀取、修改
    
        //創建節點
    /*    cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c1","c1內容".getBytes());
        cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c2","c2內容".getBytes());
        //讀取節點
        String ret1 = new String(cf.getData().forPath("/super/c2"));
        System.out.println("修改前c2節點====="+ret1);
        //修改節點
        cf.setData().forPath("/super/c2", "修改c2內容".getBytes());
        String ret2 = new String(cf.getData().forPath("/super/c2"));
        System.out.println("修改后c2節點====="+ret2);*/    
        
        
        // 綁定回調函數
        
        ExecutorService pool = Executors.newCachedThreadPool();
        cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
        .inBackground(new BackgroundCallback() {
            @Override
            public void processResult(CuratorFramework cf, CuratorEvent ce) throws Exception {
                System.out.println("code:" + ce.getResultCode());
                System.out.println("type:" + ce.getType());
                System.out.println("線程為:" + Thread.currentThread().getName());
            }
        }, pool)
        .forPath("/super/c3","c3內容".getBytes());
        Thread.sleep(Integer.MAX_VALUE);
        
        
        
        // 讀取子節點getChildren方法 和 判斷節點是否存在checkExists方法
        /**
        List<String> list = cf.getChildren().forPath("/super");
        for(String p : list){
            System.out.println(p);
        }
        
        Stat stat = cf.checkExists().forPath("/super/c3");
        System.out.println(stat);
        
        Thread.sleep(2000);
        cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");
        */
        
        
        //cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");
        cf.close();
    }
}

2、watcher方式一（監聽當前節點發生變更）

package bjsxt.curator.watcher;



import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;

import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;


public class CuratorWatcher1 {
    
    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.0.4:2181,192.168.0.9:2181,192.168.0.6: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 通過工廠創建連接
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                    .connectString(CONNECT_ADDR)
                    .sessionTimeoutMs(SESSION_OUTTIME)
                    .retryPolicy(retryPolicy)
                    .build();
        
        //3 建立連接
        cf.start();
        
        //4 建立一個cache緩存
        final NodeCache cache = new NodeCache(cf, "/super", false);
        cache.start(true);
        cache.getListenable().addListener(new NodeCacheListener() {
            /**
             * <B>方法名稱：</B>nodeChanged<BR>
             * <B>概要說明：</B>觸發事件為創建節點和更新節點，在刪除節點的時候並不觸發此操作。<BR>
             * @see org.apache.curator.framework.recipes.cache.NodeCacheListener#nodeChanged()
             */
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("監聽節點路徑=====" + cache.getCurrentData().getPath());
                System.out.println("數據=====" + new String(cache.getCurrentData().getData()));
                System.out.println("狀態=====" + cache.getCurrentData().getStat());
                System.out.println("---------------------------------------");
            }
        });
        
        Thread.sleep(1000);
        cf.create().forPath("/super", "123".getBytes());
        
        Thread.sleep(1000);
        cf.setData().forPath("/super", "456".getBytes());
        
        Thread.sleep(1000);
        cf.delete().forPath("/super");
        
        Thread.sleep(Integer.MAX_VALUE);
        
        

    }
}

3、watcher方式二（監聽當前節點的子節點發生變更）

package bjsxt.curator.watcher;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorWatcher2 {
    
    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.0.4:2181,192.168.0.9:2181,192.168.0.6: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 通過工廠創建連接
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                    .connectString(CONNECT_ADDR)
                    .sessionTimeoutMs(SESSION_OUTTIME)
                    .retryPolicy(retryPolicy)
                    .build();
        
        //3 建立連接
        cf.start();
        
        //4 建立一個PathChildrenCache緩存,第三個參數為是否接受節點數據內容 如果為false則不接受
        PathChildrenCache cache = new PathChildrenCache(cf, "/super", true);
        //5 在初始化的時候就進行緩存監聽
        cache.start(StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            /**
             * <B>方法名稱：</B>監聽子節點變更<BR>
             * <B>概要說明：</B>新建、修改、刪除<BR>
             * @see org.apache.curator.framework.recipes.cache.PathChildrenCacheListener#childEvent(org.apache.curator.framework.CuratorFramework, org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent)
             */
            @Override
            public void childEvent(CuratorFramework cf, PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                case CHILD_ADDED:
                    System.out.println("新增子節點 =====" + event.getData().getPath());
                    break;
                case CHILD_UPDATED:
                    System.out.println("修改子節點=====" + event.getData().getPath());
                    break;
                case CHILD_REMOVED:
                    System.out.println("刪除子節點=====" + event.getData().getPath());
                    break;
                default:
                    break;
                }
            }
        });

        //創建本身節點不發生變化
        cf.create().forPath("/super", "init".getBytes());
        
        //添加子節點
        Thread.sleep(1000);
        cf.create().forPath("/super/c1", "c1內容".getBytes());
        Thread.sleep(1000);
        cf.create().forPath("/super/c2", "c2內容".getBytes());
        
        //修改子節點
        Thread.sleep(1000);
        cf.setData().forPath("/super/c1", "c1更新內容".getBytes());
        
        //刪除子節點
        Thread.sleep(1000);
        cf.delete().forPath("/super/c2");        
        
        //刪除本身節點
        Thread.sleep(1000);
        cf.delete().deletingChildrenIfNeeded().forPath("/super");
        
        Thread.sleep(Integer.MAX_VALUE);
        

    }
}