zookeeper分布式集群中Curator的PathChildrenCache监测节点数据更新添加删除

野性酷女 2024-04-18 03:53 78阅读 0赞

## zookeeper分布式集群中Curator的PathChildrenCache监测节点数据更新添加删除 ##

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.retry.ExponentialBackoffRetry;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.data.Stat;
    
    import java.util.List;
    import java.util.concurrent.TimeUnit;
    
    public class Main {
        private final String TAG = "调试# ";
    
        public static void main(String[] args) {
            //初始化log4j，zookeeper否则报错。
            //org.apache.log4j.BasicConfigurator.configure();
    
            try {
                Main m = new Main();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        private String getAddress() {
            String ip = "127.0.0.1";
            return ip + ":2181," + ip + ":2182," + ip + ":2183";
        }
    
        public Main() throws Exception {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .connectString(getAddress())
                    .sessionTimeoutMs(10 * 1000)
                    .connectionTimeoutMs(20 * 1000)
                    .retryPolicy(retryPolicy)
                    .build();
    
            client.start();
    
            String path = "/test_data/my_path";
    
            //检测是否存在该路径。
            Stat stat = client.checkExists().forPath(path);
    
            //如果不存在这个路径，stat为null，创建新的节点路径。
            if (stat == null) {
                String s = client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(path);
    
                System.out.println(TAG + "已创建:" + s);
            } else {
                System.out.println(TAG + "已存在:" + path);
            }
    
            PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);
            pathChildrenCache.start();
    
            PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    System.out.println(TAG + "事件:" + event.getType());
                    System.out.println(TAG + "数据:" + event.getData().getPath() + "," + new String(event.getData().getData()));
    
                    List<ChildData> childData = event.getInitialData();
                    if (childData != null) {
                        for (ChildData data : childData) {
                            System.out.println(TAG + "子路径:" + data.getPath());
                            System.out.println(TAG + "子数据:" + data.getData().toString());
                        }
                    }
                }
            };
    
            pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
    
            //创建一个子节点。
            String tempPath = path + "/test_01";
            if (client.checkExists().forPath(tempPath) == null) {
                System.out.println(TAG + "不存在:" + tempPath);
                client.create().creatingParentsIfNeeded().forPath(tempPath, "hello,world!".getBytes("UTF-8"));
                TimeUnit.SECONDS.sleep(5);
            } else {
                System.out.println(TAG + "存在:" + tempPath);
            }
    
            update(client, tempPath);
            TimeUnit.SECONDS.sleep(5);
    
            delete(client, tempPath);
            TimeUnit.SECONDS.sleep(5);
    
            pathChildrenCache.close();
            client.close();
    
            synchronized (Main.class) {
                Main.class.wait();
            }
        }
    
        //更新操作。
        private void update(CuratorFramework client, String path) {
            //更新数据。
            try {
                client.setData()
                        //.withVersion(0)
                        .forPath(path, "zhang phil".getBytes("UTF-8"));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        //删除节点。
        private void delete(CuratorFramework client, String path) {
            try {
                client.delete().deletingChildrenIfNeeded().forPath(path);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

输出：

调试# 已创建:/test_data/my_path
    调试# 不存在:/test_data/my_path/test_01
    调试# 事件:CHILD_ADDED
    调试# 数据:/test_data/my_path/test_01,hello,world!
    调试# 事件:CHILD_UPDATED
    调试# 数据:/test_data/my_path/test_01,zhang phil
    调试# 事件:CHILD_REMOVED
    调试# 数据:/test_data/my_path/test_01,zhang phil