zoukankan      html  css  js  c++  java
  • ZooKeeper watch机制核心讲解

    ZooKeeper watch机制核心讲解

    https://blog.csdn.net/Future_LL/article/details/87700481

    Zookeeper中所有读操作(getData(),getChildren(),exists())都可以设置Watch选项。Watch事件具有one-time trigger(一次性触发)的特性,如果Watch监视的Znode有变化,那么就会通知设置该Watch的客户端。

    客户端在Znode设置了Watch时,如果Znode内容发生改变,那么客户端就会获得Watch事件。例如:客户端设置getData("/znode1", true)后,如果/znode1发生改变或者删除,那么客户端就会得到一个/znode1的Watch事件,但是/znode1再次发生变化,那客户端是无法收到Watch事件的,除非客户端设置了新的Watch。

    Watch
    ZooKeeper有watch事件,是一次性触发的【每次数据要发生变化之前都要手动创建watch】,当watch监视的数据发生时,通知设置了该watch的client,客户端即watcher。同样,其watcher是监听数据发送了某些变化,那就一定会有对应的事件类型和状态类型,一个客户端可以监控多个节点,在代码中体现在new了几个就产生几个watcher,只要节点变化都要执行一边process
    事件类型:(znode节点相关的)【针对的是你所观察的一个节点而言的】
    EventType.NodeCreated 【节点创建】
    EventType.NodeDataChanged 【节点数据发生变化】
    EventType.NodeChildrenChanged 【这个节点的子节点发生变化】
    EventType.NodeDeleted 【删除当前节点】
    状态类型:(是跟客户端实例相关的)【ZooKeeper集群跟应用服务之间的状态的变更】
    KeeperState.Disconnected 【没有连接上】
    KeeperState.SyncConnected 【连接上】
    KeeperState.AuthFailed 【认证失败】
    KeeperState.Expired 【过期】
    watcher的特性:一次性,客户端串行执行,轻量
    一次性:对于ZooKeeper的watcher,你只需要记住一点,ZooKeeper有watch事件,是一次性触发的,当watch监视的数据发生变化时,通知设置该watch的client,即watcher,由于ZooKeeper的监控都是一次性的,所以每次必须设置监控
    客户端串行执行:客户端watcher回调的过程是一个串行同步的过程,这为我们保证了顺序,同时需要开发人员注意一点,千万不要因为一个watcher的处理逻辑影响了整个客户端的watcher回调
    轻量:WatchedEvent是ZooKeeper整个Watcher通知机制的最小通知单元,整个结构只包含三个部分:通知状态、事件类型和节点路径。也就是说Watcher通知非常的简单,只会告知客户端发生了事件而不会告知其具体内容,需要客户端自己去进行获取,比如NodeDataChanged事件,ZooKeeper只会通知客户端指定节点的数据发生了变更,而不会直接提供具体的数据内容
    package bhz.zookeeper.watcher;

    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.atomic.AtomicInteger;

    import javafx.scene.shape.Path;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.EventType;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooDefs.Ids;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.data.Stat;

    /**
    * Zookeeper Wathcher
    * 本类就是一个Watcher类(实现了org.apache.zookeeper.Watcher类)
    * @author(alienware)
    * @since 2019-2-19
    */
    public class ZooKeeperWatcher implements Watcher {

    /** 定义原子变量 */
    AtomicInteger seq = new AtomicInteger();
    /** 定义session失效时间 */
    private static final int SESSION_TIMEOUT = 10000;
    /** zookeeper服务器地址 */
    private static final String CONNECTION_ADDR = "192.168.1.24:2181";
    /** zk父路径设置 */
    private static final String PARENT_PATH = "/p";
    /** zk子路径设置 */
    private static final String CHILDREN_PATH = "/p/c";
    /** 进入标识 */
    private static final String LOG_PREFIX_OF_MAIN = "【Main】";
    /** zk变量 */
    private ZooKeeper zk = null;
    /** 用于等待zookeeper连接建立之后 通知阻塞程序继续向下执行 */
    private CountDownLatch connectedSemaphore = new CountDownLatch(1);

    /**
    * 创建ZK连接
    * @param connectAddr ZK服务器地址列表
    * @param sessionTimeout Session超时时间
    */
    public void createConnection(String connectAddr, int sessionTimeout) {
    this.releaseConnection();
    try {
    //this表示把当前对象进行传递到其中去(也就是在主函数里实例化的new ZooKeeperWatcher()实例对象)
    zk = new ZooKeeper(connectAddr, sessionTimeout, this);
    System.out.println(LOG_PREFIX_OF_MAIN + "开始连接ZK服务器");
    connectedSemaphore.await();
    } catch (Exception e) {
    e.printStackTrace();
    }
    }

    /**
    * 关闭ZK连接
    */
    public void releaseConnection() {
    if (this.zk != null) {
    try {
    this.zk.close();
    } catch (InterruptedException e) {
    e.printStackTrace();
    }
    }
    }

    /**
    * 创建节点
    * @param path 节点路径
    * @param data 数据内容
    * @return
    */
    public boolean createPath(String path, String data, boolean needWatch) {
    try {
    //设置监控(由于zookeeper的监控都是一次性的所以 每次必须设置监控)
    this.zk.exists(path, needWatch);
    System.out.println(LOG_PREFIX_OF_MAIN + "节点创建成功, Path: " +
    this.zk.create( /**路径*/
    path,
    /**数据*/
    data.getBytes(),
    /**所有可见*/
    Ids.OPEN_ACL_UNSAFE,
    /**永久存储*/
    CreateMode.PERSISTENT ) +
    ", content: " + data);
    } catch (Exception e) {
    e.printStackTrace();
    return false;
    }
    return true;
    }

    /**
    * 读取指定节点数据内容
    * @param path 节点路径
    * @return
    */
    public String readData(String path, boolean needWatch) {
    try {
    System.out.println("读取数据操作...");
    return new String(this.zk.getData(path, needWatch, null));
    } catch (Exception e) {
    e.printStackTrace();
    return "";
    }
    }

    /**
    * 更新指定节点数据内容
    * @param path 节点路径
    * @param data 数据内容
    * @return
    */
    public boolean writeData(String path, String data) {
    try {
    System.out.println(LOG_PREFIX_OF_MAIN + "更新数据成功,path:" + path + ", stat: " +
    this.zk.setData(path, data.getBytes(), -1));
    } catch (Exception e) {
    e.printStackTrace();
    return false;
    }
    return true;
    }

    /**
    * 删除指定节点
    *
    * @param path
    * 节点path
    */
    public void deleteNode(String path) {
    try {
    this.zk.delete(path, -1);
    System.out.println(LOG_PREFIX_OF_MAIN + "删除节点成功,path:" + path);
    } catch (Exception e) {
    e.printStackTrace();
    }
    }

    /**
    * 判断指定节点是否存在
    * @param path 节点路径
    */
    public Stat exists(String path, boolean needWatch) {
    try {
    return this.zk.exists(path, needWatch);
    } catch (Exception e) {
    e.printStackTrace();
    return null;
    }
    }

    /**
    * 获取子节点
    * @param path 节点路径
    */
    private List<String> getChildren(String path, boolean needWatch) {
    try {
    System.out.println("读取子节点操作...");
    return this.zk.getChildren(path, needWatch);
    } catch (Exception e) {
    e.printStackTrace();
    return null;
    }
    }

    /**
    * 删除所有节点
    */
    public void deleteAllTestPath(boolean needWatch) {
    if(this.exists(CHILDREN_PATH, needWatch) != null){
    this.deleteNode(CHILDREN_PATH);
    }
    if(this.exists(PARENT_PATH, needWatch) != null){
    this.deleteNode(PARENT_PATH);
    }
    }

    /**
    * 收到来自Server的Watcher通知后的处理。
    */
    @Override
    public void process(WatchedEvent event) {

    System.out.println("进入 process 。。。。。event = " + event);

    try {
    Thread.sleep(200);
    } catch (InterruptedException e) {
    e.printStackTrace();
    }

    if (event == null) {
    return;
    }

    // 连接状态
    KeeperState keeperState = event.getState();
    // 事件类型
    EventType eventType = event.getType();
    // 受影响的path
    String path = event.getPath();
    // 原子对象seq 记录进入process的次数
    String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";

    System.out.println(logPrefix + "收到Watcher通知");
    System.out.println(logPrefix + "连接状态: " + keeperState.toString());
    System.out.println(logPrefix + "事件类型: " + eventType.toString());

    if (KeeperState.SyncConnected == keeperState) {
    // 第一次成功连接上ZK服务器
    if (EventType.None == eventType) {
    System.out.println(logPrefix + "成功连接上ZK服务器");
    connectedSemaphore.countDown();
    }
    //创建节点
    else if (EventType.NodeCreated == eventType) {
    System.out.println(logPrefix + "节点创建");
    try {
    Thread.sleep(100);
    } catch (InterruptedException e) {
    e.printStackTrace();
    }
    }
    //更新节点
    else if (EventType.NodeDataChanged == eventType) {
    System.out.println(logPrefix + "节点数据更新");
    try {
    Thread.sleep(100);
    } catch (InterruptedException e) {
    e.printStackTrace();
    }
    }
    //更新子节点
    else if (EventType.NodeChildrenChanged == eventType) {
    System.out.println(logPrefix + "子节点变更");
    try {
    Thread.sleep(3000);
    } catch (InterruptedException e) {
    e.printStackTrace();
    }
    }
    //删除节点
    else if (EventType.NodeDeleted == eventType) {
    System.out.println(logPrefix + "节点 " + path + " 被删除");
    }
    else ;
    }
    else if (KeeperState.Disconnected == keeperState) {
    System.out.println(logPrefix + "与ZK服务器断开连接");
    }
    else if (KeeperState.AuthFailed == keeperState) {
    System.out.println(logPrefix + "权限检查失败");
    }
    else if (KeeperState.Expired == keeperState) {
    System.out.println(logPrefix + "会话失效");
    }
    else ;

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

    }

    /**
    * <B>方法名称:</B>测试zookeeper监控<BR>
    * <B>概要说明:</B>主要测试watch功能<BR>
    * @param args
    * @throws Exception
    */
    public static void main(String[] args) throws Exception {

    //建立watcher //当前客户端可以称为一个watcher 观察者角色
    ZooKeeperWatcher zkWatch = new ZooKeeperWatcher();
    //创建连接
    zkWatch.createConnection(CONNECTION_ADDR, SESSION_TIMEOUT);
    //System.out.println(zkWatch.zk.toString());

    Thread.sleep(1000);

    // 清理节点
    // zkWatch.deleteAllTestPath(false);

    //-----------------第一步: 创建父节点 /p ------------------------//
    // 创建父节点
    /**
    * data:节点的内容是当前的时间毫秒值,needWATCH:是否需要监控
    */
    if (zkWatch.createPath(PARENT_PATH, System.currentTimeMillis() + "", true)) {

    Thread.sleep(1000);

    //-----------------第二步: 读取节点 /p 和 读取/p节点下的子节点(getChildren)的区别 --------------//
    // 读取数据
    // zkWatch.readData(PARENT_PATH, true);
    // 也可以这样做,表示当前的节点需要继续watch,更新数据操作就会被监控到
    zkWatch.exists(PARENT_PATH,true);

    // 读取子节点(监控childNodeChange事件)
    // 这里的监听还是监听PARENT_PATH,只是这里多了一个关于子节点的触发,这里的触发还是父节点
    // 只要子节点发生变化【删除、修改、增加】那么就触发,只会触发:NodeChildrenChanged
    zkWatch.getChildren(PARENT_PATH, true);

    // 更新数据,因为上边watch,所以更新操作会被监听到,当更新结束,上述的watch关系会消失
    zkWatch.writeData(PARENT_PATH, System.currentTimeMillis() + "");

    Thread.sleep(1000);
    // 创建子节点
    // 如果needWatch为false,只是单纯的创建了一个子节点,没有触发任何watch:NodeCreated
    zkWatch.createPath(CHILDREN_PATH, System.currentTimeMillis() + "", true);


    //-----------------第三步: 建立子节点的触发 --------------//
    // zkWatch.getChildren(CHILDREN_PATH, true);
    // zkWatch.createPath(CHILDREN_PATH + "/c1", System.currentTimeMillis() + "", true);
    // zkWatch.getChildren(CHILDREN_PATH + "/c1", true);
    // zkWatch.createPath(CHILDREN_PATH + "/c1/c2", System.currentTimeMillis() + "", true);

    //-----------------第四步: 更新子节点数据的触发 --------------//
    //在进行修改之前,我们需要watch一下这个节点:
    Thread.sleep(1000);
    zkWatch.readData(CHILDREN_PATH, true);
    zkWatch.writeData(CHILDREN_PATH, System.currentTimeMillis() + "");

    }

    Thread.sleep(10000);
    // 清理节点
    zkWatch.deleteAllTestPath(true);

    Thread.sleep(10000);
    zkWatch.releaseConnection();

    }

    ---------------------
    作者:Future_LL
    来源:CSDN
    原文:https://blog.csdn.net/Future_LL/article/details/87700481
    版权声明:本文为博主原创文章,转载请附上博文链接!

  • 相关阅读:
    中间键应用
    djano-cbv模式
    form表单使用(博客系统的登陆验证,注册)
    ajax-简介和实现注册登录
    django-settings.py配置
    Models-详细操作
    登陆验证系统实例-三种(cookie,session,auth)
    django-上传文件 fromdata(头像实例)
    学习笔记之CloudCompare
    学习笔记之Python 3
  • 原文地址:https://www.cnblogs.com/handsome1013/p/11181117.html
Copyright © 2011-2022 走看看