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);

    }

}