Zookeeper 全局循环监听节点和数据

// 一旦注册便会一直监听,只要有变化, 就会通知全局监听器
    static Watcher watcher = new Watcher() {
          
   
        @Override
        public void process(WatchedEvent watchedEvent) {
          
   
            System.out.println("状态: " + watchedEvent.getState() + ", 类型:" + watchedEvent.getType() + ",路径" + watchedEvent.getPath());
            try {
          
   
            	//连接信息
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
          
   
                    logger.info("zookeeper connect success state: {}", watchedEvent.getState());
                    long count = latch.getCount();
                    if(count == 1){
          
   
                        listenNode("/MDTService", zk, watcher);
                    }
                    latch.countDown();
                }
                if (watchedEvent.getType() == NodeDeleted) {
          
   
                    logger.info("删除节点" + watchedEvent.getPath());
                    String path = watchedEvent.getPath();
                    int i = path.lastIndexOf("/");
                    String substring = path.substring(i+1, path.length());
                    getZookeeperData(substring);
                } else if (watchedEvent.getType() == NodeCreated) {
          
   
                    logger.info("创建节点" + watchedEvent.getPath());
                } else if (watchedEvent.getType() == NodeDataChanged) {
          
   
                    logger.info("节点数据变化" + watchedEvent.getPath());
                } else if (watchedEvent.getType() == NodeChildrenChanged) {
          
   
                    logger.info("子节点变化" + watchedEvent.getPath());
                }
                if (null != watchedEvent.getPath()) {
          
   
                    Stat exists = zk.exists(watchedEvent.getPath(), false);
                    //每次调用判断是否是删除节点，如果是删除节点，则不再进行监听，不删除则再次监听
                    if (exists != null) {
          
   
                        listenNode(watchedEvent.getPath(), zk, this);
                        logger.info("{}：继续监听服务", watchedEvent.getPath());
                    } else {
          
   
                        logger.info("{}：服务停止", watchedEvent.getPath());
                    }
                }

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



	private static ZooKeeper zk = null;
    static CountDownLatch latch = new CountDownLatch(1);
    //初始化ZK连接
    public static void initZook() throws Exception {
          
   
        zk = new ZooKeeper(127.0.0.1:2181, 15000, watcher);
        latch.await();
    }

	//循环监听所有节点和所有节点内的数据
    public static void listenNode(String path, ZooKeeper zk, Watcher watcher)
            throws Exception {
          
   
        byte[] bts = zk.getData(path, watcher, new Stat());
        String str = new String(bts);
        if (str.contains("Stat")) {
          
   
            logger.info(path + ":" + str);
            String[] serverName = path.split("/");
            MonitorMsg monitor = getMonitor(serverName[serverName.length - 1], str);
            MonitorRockDB.saveMonitor(monitor);
        }
        List<String> list = zk.getChildren(path, watcher);
        if ((list.isEmpty()) || (list == null)) {
          
   
            return;
        }
        for (String s : list) {
          
   
            if (path.equals("/"))
                listenNode(path + s, zk, watcher);
            else
                listenNode(path + "/" + s, zk, watcher);
        }
    }

删除节点

添加节点，并且获取其中的数据