Zookeeper框架Curator之事件监听-yellowcong

系统运维 waitig 627℃ 百度已收录 0评论

Curator通过NodeCache、PathChildrenCache、TreeCache 对节点变化进行监听操作,以前的Zookeeper的Wacher效果差,而Curator的做法是以牺牲内存的方法来换取效率,通过NodeCache的方法,监听缓存的数据,比较缓存数据和Zookeeper服务器数据,来判断事件

1.Curator Cache 与原生ZooKeeper Wacher区别

原生的ZooKeeper Wacher是一次性的:一个Wacher一旦触发就会被移出,如果你想要反复使用Wacher,就要在Wacher被移除后重新注册,使用起来很麻烦。使用Curator Cache 可以反复使用Wacher了。

2.Curator Cache 和Curator Wacher区别

2.相关类

Curator Cache主要提供了一下三组类,分别用于实现对节点的监听,子节点的监听和二者的混合:

1.NodeCache,NodeCacheListener,ChildData,节点创建,节点数据内容变更,不能监听节点删除

2.PathChildrenCache,PathChildrenCacheListener,PathChildrenCacheEvent监听指定节点的子节点的变更包括添加,删除,子节点数据数据变更这三类。这个监听用得多

3.TreeCache,TreeCacheListener,TreeCacheEvent,TreeCacheSelector

NodeCacheListener案例

这个节点监听操作,不会监听删除事件,而且这种监听操作用到的场景很少

package com.yellowcong.zookeeper.curator;

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.recipes.cache.ChildData;
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;

/**
 * 创建日期:2017年10月14日 <br/>
 * 创建用户:yellowcong <br/>
 * 功能描述:  
 */
public class CuratorWatcherDemo {

    private static final String CONNECT_PATH = "192.168.66.110:2181,192.168.66.110:2182,192.168.66.110:2183";

    // Session 超时时间
    private static final int SESSION_TIME_OUT = 60000;

    //连接超时
    private static final int CONNECT_TIME_OUT = 5000;

    public static void main(String[] args) throws Exception {

        //1、尝试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(5000, 10);

        //2、创建CuratorFramework 工厂类
        CuratorFramework cf = CuratorFrameworkFactory.builder()
            .connectString(CONNECT_PATH)
            .connectionTimeoutMs(SESSION_TIME_OUT)
            .sessionTimeoutMs(CONNECT_TIME_OUT)
            .retryPolicy(retryPolicy)
            .build();

        //3、启动连接
        cf.start();

        //4、建立一个Cache缓存
        final NodeCache cache = new NodeCache(cf, "/curator/nodecache",false);
        cache.start();

        ExecutorService pool = Executors.newCachedThreadPool();

        //创建监听器
        cache.getListenable().addListener(new NodeCacheListener() {

            public void nodeChanged() throws Exception {
                ChildData data = cache.getCurrentData();
                System.out.println("修改路径\t"+data.getPath());
                System.out.println("数据类容\t"+new String(data.getData()));
                System.out.println("状态\t"+data.getStat());
                System.out.println("-------------------------");
            }
        },pool);

        //创建节点
        Thread.sleep(1000);
        cf.create().withMode(CreateMode.PERSISTENT).forPath("/curator/nodecache","nodecache  test".getBytes());

        //数据修改
        Thread.sleep(1000);
        cf.setData().forPath("/curator/nodecache","update".getBytes());

        //获取节点数据
        Thread.sleep(1000);
        byte [] data = cf.getData().forPath("/curator/nodecache");
        System.out.println(new String(data));


        //删除节点
        cf.delete().deletingChildrenIfNeeded().forPath("/curator/nodecache");
    }
}

PathChildrenCacheListener 案例

PathChildrenCache.StartMode
BUILD_INITIAL_CACHE //同步初始化客户端的cache,及创建cache后,就从服务器端拉入对应的数据(这个是NodeCache使用的方式)
NORMAL //异步初始化cache
POST_INITIALIZED_EVENT //异步初始化,初始化完成触发事件PathChildrenCacheEvent.Type.INITIALIZED (这个方式是 PathChildrenCacheListener 使用的)

dataIsCompressed 是否进行数据压缩 ,如果设定为false,如果在节点修改了数据,就不会直接存在缓存中,需要自己重新取一遍,一般设置为true

package com.yellowcong.zookeeper.curator;

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.recipes.cache.ChildData;
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;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

/**
 * 创建日期:2017年10月14日 <br/>
 * 创建用户:yellowcong <br/>
 * 功能描述:  
 */
public class CuratorWatcherDemo2 {

    private static final String CONNECT_PATH = "192.168.66.110:2181,192.168.66.110:2182,192.168.66.110:2183";

    // Session 超时时间
    private static final int SESSION_TIME_OUT = 60000;

    //连接超时
    private static final int CONNECT_TIME_OUT = 5000;

    public static void main(String[] args) throws Exception {

        //1、尝试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(5000, 10);

        //2、创建CuratorFramework 工厂类
        CuratorFramework cf = CuratorFrameworkFactory.builder()
            .connectString(CONNECT_PATH)
            .connectionTimeoutMs(SESSION_TIME_OUT)
            .sessionTimeoutMs(CONNECT_TIME_OUT)
            .retryPolicy(retryPolicy)
            .build();

        //3、启动连接
        cf.start();

        //4、建立一个Cache缓存 ,第三个参数是 dataIsCompressed 是否进行数据压缩  ,需要配置为true
        //如果第三个参数设置为 false,则不接受节点变更后的数据
        final PathChildrenCache cache = new PathChildrenCache(cf, "/curator", true);

        //5、设定监听的模式 ,异步初始化,初始化完成触发事件 PathChildrenCacheEvent.Type.INITIALIZED
        cache.start(StartMode.POST_INITIALIZED_EVENT);

        ExecutorService pool = Executors.newCachedThreadPool();

        //创建监听器
        cache.getListenable().addListener(new PathChildrenCacheListener() {

            public void childEvent(CuratorFramework cf, PathChildrenCacheEvent event) throws Exception {
                // TODO Auto-generated method stub
                ChildData data = event.getData();
                System.out.println("路径\t"+data.getPath());
                System.out.println("更改数据\t"+new String(data.getData()));
                System.out.println("节点状态\t"+data.getStat());
                System.out.println(event.getType());
                switch (event.getType()) {
                //初始化会触发这个事件
                case INITIALIZED:
                    System.out.println("子类缓存系统初始化完成");
                    break;
                case CHILD_ADDED:
                    System.out.println("添加子节点");
                    break;
                case CHILD_UPDATED:

                    System.out.println("更新子节点");
                    break;
                case CHILD_REMOVED:
                    System.out.println("删除子节点");
                    break;
                default:
                    break;
                }

                System.out.println("----------------------------------");
            }
        });

        //判断节点是否存在,然后删除掉
        Stat stat = cf.checkExists().forPath("/curator/nodecache2");
        if(stat != null){
            Thread.sleep(1000);
            cf.delete().deletingChildrenIfNeeded().forPath("/curator/nodecache2");
        }

        //创建节点
        Thread.sleep(1000);
        cf.create().withMode(CreateMode.PERSISTENT).forPath("/curator/nodecache2","nodecache  test".getBytes());

        //数据修改
        Thread.sleep(1000);
        cf.setData().forPath("/curator/nodecache2","update".getBytes());

        //获取节点数据
        Thread.sleep(1000);
        byte [] data = cf.getData().forPath("/curator/nodecache2");
        System.out.println(new String(data));


        //删除节点
        Thread.sleep(1000);
        cf.delete().deletingChildrenIfNeeded().forPath("/curator/nodecache2");
    }
}

本文由【waitig】发表在等英博客
本文固定链接:Zookeeper框架Curator之事件监听-yellowcong
欢迎关注本站官方公众号,每日都有干货分享!
等英博客官方公众号
点赞 (0)分享 (0)