Zookeeper watcher监听用法

zookeeper本质上是一个文件系统，目录、文件节点的创建有一定规则，这些规则应用分布式系统中，协调各节点之间的服务，常用的用法有服务发现、服务注册、服务通知等。

本文讲解监听api的用法，如何收到操作节点的通知，使用的客户端是java版本。

熟悉zookeeper客户端api的应该知道，客户端有以下几中方法，不用记住，一看便知。

    1. create，创建节点，根据持久化规则，分为永久节点和临时节点。
    1. setData，更新节点，修改节点存储的内容。
    1. getData，读取节点，读取节点存储的内容。
    1. delete，删除节点。
    1. getChildren，读取子节点内容。

创建zk客户端代码，监控节点的变化。

            ZooKeeper zk = new ZooKeeper(hostPort, sessionTimeout, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == null) {
                        return;
                    }
                    System.out.println(String.format("触发事件： [%s]， Path： [%s]", event.getType(), event.getPath()));
                }
            });

创建节点：

    private static void createNode() throws KeeperException, InterruptedException {
        zk.create("/tmp_root_path", "It is root path of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.create("/tmp_root_path/childPath1", "It is the secondary node of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.create("/tmp_root_path/childPath2", "It is the secondary node of tmp_root_path".getBytes(),  ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

执行后，会发现，客户端watcher回调并没有收到[NodeCreated]事件，这是怎么回事呢？
不妨在创建节点前，先判断节点是否存在，增强程序的健壮性。

    private static void createNodeIfNotExist() throws KeeperException, InterruptedException {
        String tmpRootPath = "/tmp_root_path";
        if (isNodeExist(tmpRootPath)) {
            System.out.println(String.format("node[%s] existed", tmpRootPath));
        } else {
            zk.create(tmpRootPath, "It is root path of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        String secondChildPath1 = "/tmp_root_path/childPath1";
        if (isNodeExist(secondChildPath1)) {
            System.out.println(String.format("node[%s] existed", secondChildPath1));
        } else {
            zk.create(secondChildPath1, "It is the secondary node of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        String secondChildPath2 = "/tmp_root_path/childPath2";
        if (isNodeExist(secondChildPath2)) {
            System.out.println(String.format("node[%s] existed", secondChildPath2));
        } else {
            zk.create(secondChildPath2, "It is the secondary node of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }
    private static boolean isNodeExist(String nodePath) {
        try {
            return zk.exists(nodePath, true) != null;
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

再次运行，有惊喜，触发通知如下：

触发事件： [NodeCreated]， Path： [/tmp_root_path]
触发事件： [NodeCreated]， Path： [/tmp_root_path/childPath1]
触发事件： [NodeCreated]， Path： [/tmp_root_path/childPath2]

很明显，三次创建节点的通知都受到了，可见，创建节点前判空方法exist是决定是否得到通知的关键，下面看exist方法，第一个参数是节点路径，第二参数就是通知客户端的关键，设置为false，发现创建节点并不会更新了。

    private static boolean isNodeExist(String nodePath) {
        try {
            return zk.exists(nodePath, false) != null;
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

更新节点

调用setData方法更新节点内容，发现并没有收到更新通知。

            zk.setData("/tmp_root_path/childPath2", "The second time updating: Child2".getBytes(), -1);

更新节点，先读取节点内容看看，有没有惊喜。

            System.out.println(new String(zk.getData("/tmp_root_path/childPath2", true, null)));
            zk.setData("/tmp_root_path/childPath2", "The second time updating: Child2".getBytes(), -1);

执行后发现，真的有惊喜，有打印客户端通知更新事件

触发事件： [NodeDataChanged]， Path： [/tmp_root_path/childPath2]

同样的，getData的第二个参数是收到通知的关键，设置为false，同样收不到更新通知。

删除节点

调用delete方法删除节点，发现并没有收到删除通知。

            zk.delete("/tmp_root_path/childPath1", -1);

同理，可推断，应该是操作节点前需要设置通知节点的动作，前置getData和isExist方法，就能收到我们想要的删除通知。

            isNodeExist("/tmp_root_path/childPath1");
            System.out.println(new String(zk.getData("/tmp_root_path/childPath1", true, null)));
            zk.delete("/tmp_root_path/childPath1", -1);

先判断节点，再读取节点，发现只会收到一次删除通知，说明重复的watcher，zk通知时已经做了过滤。

It is the secondary node of tmp_root_path
触发事件： [NodeDeleted]， Path： [/tmp_root_path/childPath1]

另外，如果删除根目录/tmp_root_path，其子目录中任然存在节点程序会报错：

org.apache.zookeeper.KeeperException$NotEmptyException: KeeperErrorCode = Directory not empty for /tmp_root_path
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:128)
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)
    at org.apache.zookeeper.ZooKeeper.delete(ZooKeeper.java:882)
    at org.wangep.websocket.dubbo.zk.TestZkWatcher.main(TestZkWatcher.java:100)

因此删除目录节点前，需要判空处理，使用getChildren方法：

            List<String> childrenList = zk.getChildren("/tmp_root_path", true);
            if (childrenList == null || childrenList.size() <= 0) {
                zk.delete("/tmp_root_path", -1);
            }

最后，最后一个带watcher的方法闪亮登场，第二个参数决定通知客户端。

总结

由上述测试可知，zk中创建、更新、删除节点前必须前置isExist、getData、getChildren方法，方法的第二个参数设置成true，只会生效一次。 getChildren()方法仅仅监控对应节点直接子目录的一次变化，但是只会监控直接子节点的增减情况，不会监控数据变化情况！若要每次对应节点发生增减变化都被监测到，那么每次都得先调用getChildren()方法获取一遍节点的子节点列表！

下面给出测试的所有代码提供参考：


public class TestZkWatcher {

    static ZooKeeper zk = null;


    private static boolean isNodeExist(String nodePath) {
        try {
            return zk.exists(nodePath, true) != null;
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }
    

    private static void createNode() throws KeeperException, InterruptedException {
        zk.create("/tmp_root_path", "It is root path of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.create("/tmp_root_path/childPath1", "It is the secondary node of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.create("/tmp_root_path/childPath2", "It is the secondary node of tmp_root_path".getBytes(),  ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    private static void createNodeIfNotExist() throws KeeperException, InterruptedException {
        String tmpRootPath = "/tmp_root_path";
        if (isNodeExist(tmpRootPath)) {
            System.out.println(String.format("node[%s] existed", tmpRootPath));
        } else {
            zk.create(tmpRootPath, "It is root path of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        String secondChildPath1 = "/tmp_root_path/childPath1";
        if (isNodeExist(secondChildPath1)) {
            System.out.println(String.format("node[%s] existed", secondChildPath1));
        } else {
            zk.create(secondChildPath1, "It is the secondary node of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        String secondChildPath2 = "/tmp_root_path/childPath2";
        if (isNodeExist(secondChildPath2)) {
            System.out.println(String.format("node[%s] existed", secondChildPath2));
        } else {
            zk.create(secondChildPath2, "It is the secondary node of tmp_root_path".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }


    public static void main(String[] args) {

        try {
            System.out.println("Starting to connect Zk......");
            String hostPort = "localhost:2181";
            int sessionTimeout = 5000;

            zk = new ZooKeeper(hostPort, sessionTimeout, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == null) {
                        return;
                    }
                    System.out.println(String.format("触发事件： [%s]， Path： [%s]", event.getType(), event.getPath()));
                }
            });

            System.out.println("Zookeeper connected");

            TimeUnit.SECONDS.sleep(1);

            createNodeIfNotExist();


//            System.out.println(new String(zk.getData("/tmp_root_path/childPath2", true, null)));
            isNodeExist("/tmp_root_path/childPath2");
            zk.setData("/tmp_root_path/childPath2", "The second time updating: Child2".getBytes(), -1);

            isNodeExist("/tmp_root_path/childPath1");
            System.out.println(new String(zk.getData("/tmp_root_path/childPath1", true, null)));
            zk.delete("/tmp_root_path/childPath1", -1);
            zk.delete("/tmp_root_path/childPath2", -1);

            List<String> childrenList = zk.getChildren("/tmp_root_path", true);
            if (childrenList == null || childrenList.size() <= 0) {
                zk.delete("/tmp_root_path", -1);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 194,088评论 5赞 459
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 81,715评论 2赞 371
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 141,361评论 0赞 319
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 52,099评论 1赞 263
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 60,987评论 4赞 355
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 46,063评论 1赞 272
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 36,486评论 3赞 381
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 35,175评论 0赞 253
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 39,440评论 1赞 290
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,518评论 2赞 309
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,305评论 1赞 326
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 32,190评论 3赞 312
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 37,550评论 3赞 298
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 28,880评论 0赞 17
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,152评论 1赞 250
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 41,451评论 2赞 341
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 40,637评论 2赞 335