Zookeeper使用记录

@Component
public class ZookeeperClientConfig {

    private CuratorFramework client;

    @PostConstruct
    public void init() {
        client = createCuratorFramework("127.0.0.1:2181");
        client.start();
    }

    public CuratorFramework createCuratorFramework(String connectString) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().connectString(connectString)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(30000)
                .sessionTimeoutMs(30000)
                .build();
        return curatorFramework;
    }

    @Bean
    public CuratorFramework zookeeperClient() {
        return client;
    }
}

@Test
public void testOperation() throws Exception {
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    CuratorFramework client = CuratorFrameworkFactory.builder().connectString("192.168.56.101:2181")
            .retryPolicy(retryPolicy)
            .connectionTimeoutMs(30000)
            .sessionTimeoutMs(30000)
            .namespace("test")
            .build();

    client.start();
    //创建永久节点
    client.create().forPath("/test", "/test data".getBytes());

    //创建永久有序节点
    client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/test_sequential", "/test_sequential data".getBytes());

    //创建临时节点
    client.create().withMode(CreateMode.EPHEMERAL)
            .forPath("/test/ephemeral", "/test/ephemeral data".getBytes());

    //创建临时有序节点
    client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/test/ephemeral_path1", "/test/ephemeral_path1 data".getBytes());

    client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/test/ephemeral_path2", "/test/ephemeral_path2 data".getBytes());

    //测试检查某个节点是否存在
    Stat stat1 = client.checkExists().forPath("/test");
    Stat stat2 = client.checkExists().forPath("/test2");

    System.out.println("'/test'是否存在： " + (stat1 != null));
    System.out.println("'/test2'是否存在： " + (stat2 != null));

    //获取某个节点的所有子节点
    System.out.println(client.getChildren().forPath("/"));

    //获取某个节点数据
    System.out.println(new String(client.getData().forPath("/test")));

    //设置某个节点数据
    client.setData().forPath("/test", "/test modified data".getBytes());

    //创建测试节点
    client.create().orSetData().creatingParentContainersIfNeeded()
            .forPath("/test/del_key1", "/test/del_key1 data".getBytes());
    client.create().orSetData().creatingParentContainersIfNeeded()
            .forPath("/test/del_key2", "/test/del_key2 data".getBytes());

    client.create().forPath("/test/del_key2/test_key", "test_key data".getBytes());

    //删除该节点
    client.delete().forPath("/test/del_key1");

    //级联删除子节点
    client.delete().guaranteed().deletingChildrenIfNeeded().forPath("/test/del_key2");
}

@Test
public void testPathChildrenCache() throws Exception{
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181")
            .retryPolicy(retryPolicy)
            .connectionTimeoutMs(30000)
            .sessionTimeoutMs(30000)
            //.namespace("test")
            .build();

    client.start();

    PathChildrenCache cache = new PathChildrenCache(client, "/test", true);

    cache.getListenable().addListener(new PathChildrenCacheListener() {
        @Override
        public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
            if(pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)){
                log.info("子节点初始化成功");
            }else if(pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)){
                log.info("添加子节点路径:"+pathChildrenCacheEvent.getData().getPath());
                if (pathChildrenCacheEvent.getData() != null) {
                    log.info("子节点数据:{}", pathChildrenCacheEvent.getData().getData());
                }
            }else if(pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
                log.info("删除子节点:"+pathChildrenCacheEvent.getData().getPath());
            }else if(pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
                log.info("修改子节点路径:"+pathChildrenCacheEvent.getData().getPath());
                if (pathChildrenCacheEvent.getData() != null) {
                    log.info("修改子节点数据:{}", pathChildrenCacheEvent.getData().getData());
                }
            }
        }
    });
    /*
        * StartMode：初始化方式
        * POST_INITIALIZED_EVENT：异步初始化。初始化后会触发事件
        * NORMAL：异步初始化
        * BUILD_INITIAL_CACHE：同步初始化
        *
        * */
    cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
}

@Test
public void testLock() throws Exception {
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181")
            .retryPolicy(retryPolicy)
            .connectionTimeoutMs(30000)
            .sessionTimeoutMs(30000)
            .namespace("test")
            .build();

    client.start();

    InterProcessMutex lock = new InterProcessMutex(client, "/lock");
    lock.acquire();
    Thread.sleep(10000);
    lock.release();

    client.close();
}