public class ZkClientCrud<T> {

    ZkClient zkClient ;
    final static Logger logger = LoggerFactory.getLogger(ZkClientCrud.class);

    public ZkClientCrud(ZkSerializer zkSerializer) {
        logger.info("链接zk开始");
        // zkClient=new ZkClient(ZookeeperUtil.connectString,ZookeeperUtil.sessionTimeout);
        zkClient=new ZkClient(ZookeeperUtil.connectString,ZookeeperUtil.sessionTimeout,ZookeeperUtil.sessionTimeout,zkSerializer);
    }


    public void createEphemeral(String path,Object data){

        zkClient.createEphemeral(path,data);
    }

    /***
     * 支持创建递归方式
     * @param path
     * @param createParents
     */
    public void createPersistent(String path,boolean createParents){

        zkClient.createPersistent(path,createParents);
    }

    /***
     * 创建节点 跟上data数据
     * @param path
     * @param data
     */
    public void createPersistent(String path,Object data){

        zkClient.createPersistent(path,data);
    }

    /***
     * 子节点
     * @param path
     * @return
     */
    public  List<String> getChildren(String path){
        return zkClient.getChildren(path);

    }

    public  T readData(String path){
        return zkClient.readData(path);
    }

    public  void  writeData(String path,Object data){
        zkClient.writeData(path,data);
    }

    //递归删除
    public  void deleteRecursive(String path){
        zkClient.deleteRecursive(path);

    }
}

public class ZkClientWatcher {
    ZkClient zkClient;

    public ZkClientWatcher() {
        zkClient = new ZkClient(new ZkConnection(ZookeeperUtil.connectString), ZookeeperUtil.sessionTimeout);
    }
    
    public void createPersistent(String path, Object data) {
        zkClient.createPersistent(path, data);
    }


    public void writeData(String path, Object object) {
        zkClient.writeData(path, object);

    }

    public void delete(String path) {
        zkClient.delete(path);

    }

    public boolean exists(String path) {
        return zkClient.exists(path);

    }

    public void deleteRecursive(String path) {
        zkClient.deleteRecursive(path);

    }

    //对父节点添加监听数据变化。
    public void subscribe(String path) {


        zkClient.subscribeDataChanges(path, new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                System.out.printf("变更的节点为:%s,数据：%s\r\n", dataPath, data);
            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.printf("删除的节点为:%s\r\n", dataPath);
            }
        });
    }

    //对父节点添加监听子节点变化。
    public void subscribe2(String path) {
        zkClient.subscribeChildChanges(path, new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                System.out.println("父节点: " + parentPath + ",子节点:" + currentChilds + "\r\n");
            }
        });
    }
    
    //客户端状态
    public void subscribe3(String path) {
        zkClient.subscribeStateChanges(new IZkStateListener() {
            @Override
            public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception {
                if (state == Watcher.Event.KeeperState.SyncConnected) {
                    //当我重新启动后start，监听触发
                    System.out.println("连接成功");
                } else if (state == Watcher.Event.KeeperState.Disconnected) {
                    System.out.println("连接断开");//当我在服务端将zk服务stop时，监听触发
                } else
                    System.out.println("其他状态" + state);
            }

            @Override
            public void handleNewSession() throws Exception {
                System.out.println("重建session");

            }

            @Override
            public void handleSessionEstablishmentError(Throwable error) throws Exception {

            }
        });

    }

  /*  @Override
    public void handleDataChange(String dataPath, Object data) throws Exception {
    }

    @Override
    public void handleDataDeleted(String dataPath) throws Exception {

    }*/
}

public class CuratorDao {

    //使用curator(递归)添加节点
    //级联创建节点（原生API不支持/后台客户端也不支持，但是Curator支持）
    public static void createNodes(CuratorFramework zkClient, String nodePath, String nodeData) throws Exception {
        zkClient.create()
                .creatingParentContainersIfNeeded()//创建父节点,如果需要的话
                .withMode(CreateMode.PERSISTENT) //指定节点是临时的，还是永久的
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE) //指定节点的操作权限
                .forPath(nodePath, nodeData.getBytes());
        System.out.println(nodePath + "节点已成功创建…");
    }

    //使用curator(递归)删除节点
    //删除node节点及其子节点
    public static void deleteNodeWithChild(CuratorFramework zkClient, String nodePath) throws Exception {
        zkClient.delete()
                .guaranteed()                //保证删除：如果删除失败，那么在后端还是继续会删除，直到成功
                .deletingChildrenIfNeeded()  //级联删除子节点
                //.withVersion(1)//版本号可以据需使用
                .forPath(nodePath);
        System.out.println(nodePath + "节点已删除成功…");
    }

    //使用curator更新节点数据
    //更新节点data数据
    public static void updateNodeData(CuratorFramework zkClient, String nodePath, String nodeNewData) throws Exception {
        zkClient.setData().withVersion(0).forPath(nodePath, nodeNewData.getBytes());//版本号据需使用，默认可以不带
        System.out.println(nodePath + "节点数据已修改成功…");
    }

    //使用curator查询节点数据
    //查询node节点数据
    public static void getNodeData(CuratorFramework zkClient, String nodePath) throws Exception {
        Stat stat = new Stat();
        byte[] data = zkClient.getData().storingStatIn(stat).forPath(nodePath);
        System.out.println("节点" + nodePath + "的数据为" + new String(data));
        System.out.println("节点的版本号为：" + stat.getVersion());
    }

    //使用curator查询节点的子节点
    //打印node子节点
    public static void printChildNodes(CuratorFramework zkClient, String parentNodePath) throws Exception {
        List<String> childNodes = zkClient.getChildren().forPath(parentNodePath);
        System.out.println("开始打印子节点");
        for (String str : childNodes) {
            System.out.println(str);
        }
    }

    //使用curator判断节点是否存在
    //判断node节点是否存在
    public static void checkNodeExists(CuratorFramework zkClient, String nodePath) throws Exception {
        Stat stat = zkClient.checkExists().forPath(nodePath);
        System.out.println(null == stat ? "节点不存在" : "节点存在");
    }

    /**************使用Curator高级API特性之Cache缓存监控节点变化*************/


    @Test
    public void test() throws Exception {
        ZkConnectCuratorUtil zkUtil = new ZkConnectCuratorUtil();
        CuratorFramework zkClient = zkUtil.zkClient;
//        CuratorDao.createNodes(zkClient,"/xiaosi/test","siguogui");
//        CuratorDao.deleteNodeWithChild(zkClient,"/xiaosi/test");
//        CuratorDao.updateNodeData(zkClient,"/xiaosi/test","xiaosi");
//        CuratorDao.getNodeData(zkClient,"/xiaosi/test");
//        CuratorDao.printChildNodes(zkClient, "/xiaosi");
        CuratorDao.checkNodeExists(zkClient, "/xiaosi");
    }

}

package org.example.zookeeper.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import org.testng.annotations.Test;

import java.util.List;

/**
 * @ClassName: CuratorDao
 * @Description:
 * @Author: 88578
 * @Date: 2022/5/1 14:17
 */
public class CuratorDao {

    //使用curator(递归)添加节点
    //级联创建节点（原生API不支持/后台客户端也不支持，但是Curator支持）
    public static void createNodes(CuratorFramework zkClient, String nodePath, String nodeData) throws Exception {
        zkClient.create()
                .creatingParentContainersIfNeeded()//创建父节点,如果需要的话
                .withMode(CreateMode.PERSISTENT) //指定节点是临时的，还是永久的
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE) //指定节点的操作权限
                .forPath(nodePath, nodeData.getBytes());
        System.out.println(nodePath + "节点已成功创建…");
    }

    //使用curator(递归)删除节点
    //删除node节点及其子节点
    public static void deleteNodeWithChild(CuratorFramework zkClient, String nodePath) throws Exception {
        zkClient.delete()
                .guaranteed()                //保证删除：如果删除失败，那么在后端还是继续会删除，直到成功
                .deletingChildrenIfNeeded()  //级联删除子节点
                //.withVersion(1)//版本号可以据需使用
                .forPath(nodePath);
        System.out.println(nodePath + "节点已删除成功…");
    }

    //使用curator更新节点数据
    //更新节点data数据
    public static void updateNodeData(CuratorFramework zkClient, String nodePath, String nodeNewData) throws Exception {
        zkClient.setData().withVersion(0).forPath(nodePath, nodeNewData.getBytes());//版本号据需使用，默认可以不带
        System.out.println(nodePath + "节点数据已修改成功…");
    }

    //使用curator查询节点数据
    //查询node节点数据
    public static void getNodeData(CuratorFramework zkClient, String nodePath) throws Exception {
        Stat stat = new Stat();
        byte[] data = zkClient.getData().storingStatIn(stat).forPath(nodePath);
        System.out.println("节点" + nodePath + "的数据为" + new String(data));
        System.out.println("节点的版本号为：" + stat.getVersion());
    }

    //使用curator查询节点的子节点
    //打印node子节点
    public static void printChildNodes(CuratorFramework zkClient, String parentNodePath) throws Exception {
        List<String> childNodes = zkClient.getChildren().forPath(parentNodePath);
        System.out.println("开始打印子节点");
        for (String str : childNodes) {
            System.out.println(str);
        }
    }

    //使用curator判断节点是否存在
    //判断node节点是否存在
    public static void checkNodeExists(CuratorFramework zkClient, String nodePath) throws Exception {
        Stat stat = zkClient.checkExists().forPath(nodePath);
        System.out.println(null == stat ? "节点不存在" : "节点存在");
    }

    /**************使用Curator高级API特性之Cache缓存监控节点变化*************/

    //一次性的watch
    public static void watchOnce(CuratorFramework zkClient, String nodePath) throws Exception {
        zkClient.getData().usingWatcher(new MyCuratorWatcher()).forPath(nodePath);
    }

    //NodeCache监听当前节点变化
    //通过NodeCacheListener接口持续监听节点的变化来实现
    //持续监听的watch
    public static void watchForeverByNodeCache(CuratorFramework zkClient, String nodePath) throws Exception {
        final NodeCache nodeCache = new NodeCache(zkClient, nodePath);//把监听节点，转换为nodeCache
        nodeCache.start(false);//默认为false  设置为true时，会自动把节点数据存放到nodeCache中；设置为false时，初始化数据为空
        ChildData cacheData = nodeCache.getCurrentData();
        if (null == cacheData) {
            System.out.println("NodeCache节点的初始化数据为空……");
        } else {
            System.out.println("NodeCache节点的初始化数据为" + new String(cacheData.getData()));
        }

        //设置循环监听
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                ChildData cdata = nodeCache.getCurrentData();
                if (null == cdata) {
                    System.out.println("节点发生了变化，可能刚刚被删除！");
                    nodeCache.close();//关闭监听
                } else {
                    String data = new String(cdata.getData());
                    String path = nodeCache.getCurrentData().getPath();
                    System.out.println("节点路径" + path + "数据发生了变化，最新数据为：" + data);
                }
            }
        });
    }

    //PathChildrenCache只监听子节点变化
    //通过PathChildrenCacheListener接口持续监听子节点来实现
    //持续监听watch子节点的任何变化
    public static void watchForeverByPathChildrenCache(CuratorFramework zkClient, String nodePath) throws Exception {
        final PathChildrenCache childrenCache = new PathChildrenCache(zkClient, nodePath, true);//把监听节点，转换为childrenCache
        /**
         * StartMode:初始化方式
         *	POST_INITIALIZED_EVENT: 异步初始化，初始化之后会触发事件(会进入下面的第一个case)
         *	NORMAL:异步初始化 (不会进入下面的第一个case)
         *	BUILD_INITIAL_CACHE: 同步初始化(把节点数据同步缓存到Cache中)
         */
        childrenCache.start(PathChildrenCache.StartMode.NORMAL);
        List<ChildData> childDataList = childrenCache.getCurrentData();
        System.out.println("当前节点所有子节点的数据列表如下：");
        for (ChildData childData : childDataList) {
            System.out.println(new String(childData.getData()));
        }

        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {

            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case INITIALIZED:
                        System.out.println("子节点初始化OK…");
                        break;
                    case CHILD_ADDED:
                        System.out.println("子节点" + event.getData().getPath() + "已被成功添加，数据data=" + new String(event.getData().getData()));
                        break;
                    case CHILD_UPDATED:
                        System.out.println("子节点" + event.getData().getPath() + "数据发生变化，新数据data=" + new String(event.getData().getData()));
                        break;
                    case CHILD_REMOVED:
                        System.out.println("子节点" + event.getData().getPath() + "已被移除~");
                        break;
                    case CONNECTION_RECONNECTED:
                        System.out.println("正在尝试重新建立连接…");
                        break;
                    case CONNECTION_SUSPENDED:
                        System.out.println("连接状态被暂时停止…");
                        break;
                    default:
                        break;
                }
            }
        });
    }

    //TreeCache是上两者的合体，既监听自身，也监听所有子节点变化
    //通过TreeCacheListener接口来实现
    public static void treeCache(CuratorFramework zkClient, String nodePath) throws Exception {
//        final String path = "/treeChildrenCache";
        final TreeCache treeCache = new TreeCache(zkClient, nodePath);

        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case NODE_ADDED:
                        System.out.println("节点变动触发：NODE_ADDED:" + event.getData().getPath());
                        break;
                    case NODE_REMOVED:
                        System.out.println("节点变动触发：NODE_REMOVED:" + event.getData().getPath());
                        break;
                    case NODE_UPDATED:
                        System.out.println("节点变动触发：NODE_UPDATED:" + event.getData().getPath());
                        break;
                    case CONNECTION_LOST:
                        System.out.println("节点变动触发：CONNECTION_LOST:" + event.getData().getPath());
                        break;
                    case CONNECTION_RECONNECTED:
                        System.out.println("节点变动触发：CONNECTION_RECONNECTED:" + event.getData().getPath());
                        break;
                    case CONNECTION_SUSPENDED:
                        System.out.println("节点变动触发：CONNECTION_SUSPENDED:" + event.getData().getPath());
                        break;
                    case INITIALIZED:
                        System.out.println("节点变动触发：INITIALIZED:" + event.getData().getPath());
                        break;
                    default:
                        break;
                }
            }
        });
        //据需可以继续做一些其他的增删改操作
        zkClient.create().withMode(CreateMode.PERSISTENT).forPath(nodePath);
        Thread.sleep(1000);

        zkClient.create().withMode(CreateMode.PERSISTENT).forPath(nodePath + "/c1");
        Thread.sleep(1000);

        zkClient.delete().forPath(nodePath + "/c1");
        Thread.sleep(1000);

        zkClient.delete().forPath(nodePath);
        Thread.sleep(1000);
        zkClient.close();
    }

    /*Curator Cache，是在zk3.6新版本添加的特性，Curator需5.*+
    它的出现是为了替换以上3个监听（NodeCache、PathCache、TreeCache），
            它通过CuratorCacheListener.builder().for***来选择对应的监听。
            最后再通过curatorCache.listenable().addListener(listener);注册监听。*/
    public static void curatorCache1(CuratorFramework zkClient) {
        final String path = "/curatorCache";
        CuratorCache curatorCache = CuratorCache.build(zkClient, path);
        curatorCache.listenable().addListener(new CuratorCacheListener() {
            @Override
            public void event(Type type, ChildData oldData, ChildData newdata) {
                switch (type) {
                    case NODE_CREATED:
                        //各种判断
                        break;

                    default:
                        break;
                }
            }
        });
    }

    public static void curatorCache2(CuratorFramework zkClient) throws InterruptedException {
        final String path = "/curatorCache";
        CuratorCache curatorCache = CuratorCache.builder(zkClient, path).build();

        //构建监听器
        //新旧对照：
        //1.node cache--> CuratorCacheListener.builder().forNodeCache(new NodeCacheListener(){} );
        //2.path cache--> CuratorCacheListener.builder().forPathChildrenCache();
        //3.tree cache--> CuratorCacheListener.builder().forTreeCache.forTreeCache();
        CuratorCacheListener listener = CuratorCacheListener.builder()
                .forNodeCache(new NodeCacheListener() {
                    @Override
                    public void nodeChanged() throws Exception {
                        System.out.println("节点改变了...");
                    }
                })
                .build();

        //添加监听
        curatorCache.listenable().addListener(listener);
        //开启监听
        curatorCache.start();
        //让线程休眠30s(为了方便测试)
        Thread.sleep(1000 * 30);
    }


    @Test
    public void test() throws Exception {
        ZkConnectCuratorUtil zkUtil = new ZkConnectCuratorUtil();//new的同时，zk也被启动
        CuratorFramework zkClient = zkUtil.zkClient;
//        CuratorDao.createNodes(zkClient,"/xiaosi/test","siguogui");
//        CuratorDao.deleteNodeWithChild(zkClient,"/xiaosi/test");
//        CuratorDao.updateNodeData(zkClient,"/xiaosi/test","xiaosi");
//        CuratorDao.getNodeData(zkClient,"/xiaosi/test");
//        CuratorDao.printChildNodes(zkClient, "/xiaosi");
        CuratorDao.checkNodeExists(zkClient, "/xiaosi");
    }


    public static void main(String[] args) throws Exception {
        ZkConnectCuratorUtil zkUtil = new ZkConnectCuratorUtil();//new的同时，zk也被启动
        CuratorFramework zkClient = zkUtil.zkClient;
//        CuratorDao.watchOnce(zkClient, "/xiaosi/test");
//        CuratorDao.watchForeverByNodeCache(zkClient, "/xiaosi/test");
//        CuratorDao.watchForeverByPathChildrenCache(zkClient, "/xiaosi/test");
        CuratorDao.treeCache(zkClient, "/xiaosi/test4");

        CuratorDao dao = new CuratorDao();

        String nodePath = "/super/succ";
        dao.createNodes(zkClient, nodePath, "super");//创建节点
//		dao.updateNodeData(zkClient, nodePath, "hello");//更新节点数据
//		dao.deleteNodeWithChild(zkClient, nodePath);
//		dao.getNodeData(zkClient, nodePath);
//		dao.printChildNodes(zkClient, nodePath);
//		dao.checkNodeExists(zkClient, nodePath);
//		dao.watchOnce(zkClient, nodePath);
//		dao.watchForeverByNodeCache(zkClient, nodePath);
//		dao.watchForeverByPathChildrenCache(zkClient, nodePath);

        Thread.sleep(300000); //延迟sleep时间，便于后才修改节点，看前台是否会继续触发watch
        cto.closeZKClient();
    }
}

public class ZkLock {
    final static Logger log = LoggerFactory.getLogger(ZkLock.class);
    public CuratorFramework zkClient = null; // zk的客户端工具Curator（在本类通过new实例化的是，自动start）
    private static final int BASE_SLEEP_TIME_MS = 1000; // 连接失败后，再次重试的间隔时间 单位:毫秒
    private static final int MAX_RETRY_TIMES = 10; // 定义失败重试次数
    private static final int SESSION_TIME_OUT = 1000000; // 会话存活时间,根据业务灵活指定 单位:毫秒
    private static final String ZK_SERVER_IP_PORT = "localhost:2181";// Zookeeper服务所在的IP和客户端端口
    private static final String NAMESPACE = "workspace";// 指定后，默认操作的所有的节点都会在该工作空间下进行
    static int j = 10;

    //初始化zk客户端
    public ZkLock() {
        // 重试策略：初试时间为1s 重试10次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME_MS, MAX_RETRY_TIMES);
        // 通过工厂建立连接
        zkClient = CuratorFrameworkFactory.builder().connectString(ZK_SERVER_IP_PORT) // 连接地址
                .sessionTimeoutMs(SESSION_TIME_OUT).retryPolicy(retryPolicy)// 重试策略
                .build();
        zkClient.start();
    }

    public static void lockTest(CuratorFramework zkClient) throws InterruptedException {
        // 使用分布式锁,所有系统同时监听同一个节点,达到分布式锁的目的
        final InterProcessMutex lock = new InterProcessMutex(zkClient, "/test");
        final CountDownLatch countDownLatch = new CountDownLatch(1);
        for (int i = 0; i < 10; i++) {//启动10个线程
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        countDownLatch.await();// 线程等待一起执行
                        lock.acquire();// 分布式锁,数据同步
                        // 处理业务
                        j--;
                        System.out.println(j);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        try {// 释放锁
                            lock.release();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }, "t" + i).start();
        }
        Thread.sleep(1000);
        countDownLatch.countDown();// 模拟十个线程一起并发.指定一起执行
    }

    public static void main(String[] args) throws InterruptedException {
        ZkLock zkl = new ZkLock();
        ZkLock.lockTest(zkl.zkClient);
    }
}