zookeeper的监听器原理

Posted 2021-09-09 阳哥赚钱很牛

        客户端注册监听它关心的目录节点，当目录节点发生变化（数据改变、节点删除、子目 录节点增加删除）时，ZooKeeper 会通知客户端。监听机制保证 ZooKeeper 保存的任何的数据的任何改变都能快速的响应到监听了该节点的应用程序。

        zookeeper只负责通知，具体做不做还要看客户端

1、监听原理详解

（1）首先要有一个main()线程

（2）在main线程中创建Zookeeper客户端，这时就会创建两个线程，一个负责网络连接通信(connet)，一个负责监听(listener) 。

（3）通过connect线程将注册的监听事件发送给Zookeeper。

（4）在Zookeeper的注册监听器列表中将注册的监听事件添加到列表中。

（5）Zookeeper监听到有数据或路径变化，就会将这个消息发送给listener线程。

（6） listener线程内部调用了process()方法。

2、常见的监听

（1）监听节点数据的变化

get path [ watch]

（2）监听子节点增减的变化

ls path [watch]

一、节点的值变化监听

（1）在 node1 主机上注册监听/client1节点数据变化

[zk: localhost:2181(CONNECTED) 26] get -w /client1

（2）在node2主机上修改/client1节点的数据

[zk: localhost:2181(CONNECTED) 1] set /client1 "xisi"

（3）观察 node1主机收到数据变化的监听

WATCHER:

WatchedEvent state:SyncConnected

type:NodeDataChanged

path:/client1

注意：在node2再多次修改/client1的值，node上不会再收到监听。因为注册一次，只能监听一次。想再次监听，需要再次注册。

二、节点的子节点变化监听（路径变化）

（1）在 node1主机上注册监听/client1节点的子节点变化

[zk: localhost:2181(CONNECTED) 1] ls -w /client1

[yg, tt]

（2）在 node2 主机/client2 节点上创建子节点

[zk: localhost:2181(CONNECTED) 2] create /client1/shizi "shizi"

Created /client1/shizi

（3）观察 node1 主机收到子节点变化的监听

WATCHER::

WatchedEvent state:SyncConnected type:NodeChildrenChanged

path:/client1

注意：节点的路径变化，也是注册一次，生效一次。想多次生效，就需要多次注册。

 3、zookeeper的监听器的JAVA API

        zookeeper中存储的节点以及节点信息都可以被客户端监听，一旦客户端监听的自己感兴趣的事情产生，那么zookeeper就会通知客户端去做出一些反应。

连接zookeeper一共有三个参数：
        1、String connectStr ：zookeeper的地址
        2、sessionTimeout：连接超时时长
        3、watcher：zookeeper的监听器

监听器触发有如下几种情况：
        1、通过客户端去查询或者创建、修改zookeeper上的节点都会去触发一次监听器
        2、再去查询zookeeper的节点数据、子节点、节点存在与否的时候也会注册监听器，那么此时这个监听器也可以让他们三个取用

public class ZookeeperWatcherTest {
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        /**
         *
         *关心的监听器有四个地方可以触发：
         *    1、构建zookeeper对象之后 调用zookeeper的方法时候监听器会被触发---不常用
         *    2、getData()---获取节点数据的  里面注册监听器
         *         被触发--当节点的数据发生变化的时候
         *    3、getChildren()----获取节点的子节点信息
         *         被触发---当前节点的子节点列表发生变化会触发
         *    4、exists() ----判断节点存不存在
         *         被触发---当节点被创建或者删除会触发
         *                 或者当节点的数据变化也会去触发
         */
        ZooKeeper zkClient = new ZooKeeper("node1:2181,node2:2181,node3:2181", 2000, new Watcher() {
//Zookeeper()方法中第一个值代表着zookeeper的地址，第二个值代表zookeeper连接超时时间，最后一个是监听器
            @Override
            public void process(WatchedEvent event) {
                System.out.println("zkClient的监听器被触发了");
            }
        });
//        byte[] data = zkClient.getData("/java", true, new Stat());//true代表使用zookeeper监听器
        byte[] data = zkClient.getData("/java", new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("getData自定义的监听器被触发了");
            }
        }, new Stat());
        System.out.println(new String(data));
        Thread.sleep(Long.MAX_VALUE);
    }
}

监听器有四个地方可以触发：
        1、构建zookeeper对象之后 调用zookeeper的方法时候监听器会被触发---不常用
        2、getData()---获取节点数据的  里面注册监听器
                被触发--当节点的数据发生变化的时候
         3、getChildren()----获取节点的子节点信息
                被触发---当前节点的子节点列表发生变化会触发
         4、exists() ----判断节点存不存在
                被触发---当节点被创建或者删除会触发
                或者当节点的数据变化也会去触发

如何做到持久化监听呢？

public class MoreTimesWatcher {
    static  ZooKeeper zkClient;
    public static void main(String[] args) throws Exception {
        zkClient= new ZooKeeper("node1:2181,node2:2181,node3:2181", 2000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                /**
                 * 如果想持久监听 我们需要再监听器被触发之后再次去注册一下这个监听器即可
                 */
                try {
                    byte[]  data = zkClient.getData("/java", true, new Stat());
                    System.out.println(new String(data));
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        });
        byte[] data = zkClient.getData("/java", true, new Stat());//监听一次以后再去执行process()方法重新调用此监听器进行监听
        System.out.println(new String(data));
        Thread.sleep(Long.MAX_VALUE);
    }
}

接下来我们看一下子节点列表监听的java代码：

/**
 * 监听子节点列表的变化
 */
public class NodeListWatcher {
    static ZooKeeper zkClient;
    public static void main(String[] args) throws Exception {
        zkClient= new ZooKeeper("node1:2181,node2:2181,node3:2181", 2000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                try {
                    List<String> children = zkClient.getChildren("/java", true);
                    System.out.println(children);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        List<String> children = zkClient.getChildren("/java", true);
        System.out.println(children);
        Thread.sleep(Long.MAX_VALUE);
    }
}

判断节点是否存在：

/**
 * 判断节点是否存在的一个监听器：
 *    两种情况：
 *      1、节点创建或删除
 *      2、节点数据发生变化也会触发
 */
public class ExistsWatcher {
    static ZooKeeper zkClient;
    public static void main(String[] args) throws Exception {
        zkClient= new ZooKeeper("node1:2181,node2:2181,node3:2181", 2000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                try {
                    Stat stat  = zkClient.exists("/java", true);
                    System.out.println(stat==null?"节点不存在":"节点存在");
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        Stat stat = zkClient.exists("/java", true);
        System.out.println(stat==null?"节点不存在":"节点存在");
        Thread.sleep(Long.MAX_VALUE);
    }
}