Zookeeper客户端Curator使用(增删改查)

说明

注意: 如未安装Zookeeper服务端，请先移步《docker安装Zookeeper(单点)》搭建简易zk环境。

本篇文章关于Zookeeper客户端Curator使用示例。

简介

Apache Curator 是一个用于Apache ZooKeeper 的Java 客户端框架。 ZooKeeper 是一个分布式协调服务，用于在分布式系统中进行协作和管理。 Curator 提供了一组易于使用的API和工具，简化了与ZooKeeper 的交互，同时提供了更高级别的抽象和功能。

Curator框架主要解决了三类问题：

• 封装ZooKeeper Client与ZooKeeper Server之间的连接处理（提供连接重试机制等）。

• 提供了一套Fluent风格的API，并且在Java客户端原生API的基础上进行了增强（创捷多层节点、删除多层节点等）。

• 提供ZooKeeper各种应用场景（分布式锁、leader选举、共享计数器、分布式队列等）的抽象封装。

使用

1.引入Curator

<dependency>
      <groupId>org.apache.curator</groupId>
      <artifactId>curator-framework</artifactId>
      <version>5.4.0</version>
</dependency>

<dependency>
      <groupId>org.apache.curator</groupId>
      <artifactId>curator-recipes</artifactId>
      <version>5.4.0</version>
</dependency>

2.建立连接

    /**
    * 建立zk连接
    *
    * @param connString zk连接字符串, 如"192.169.7.171:2181,192.169.7.172:2181,192.169.7.173:2181"
    * @return CuratorFramework
    */
    public static CuratorFramework getConn1(String connString) {
        //重试策略 baseSleepTimeMs 重试之间等待的初始时间，maxRetries 重试的最大次数
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,5);

        return CuratorFrameworkFactory.newClient(connString, 60 * 1000, 15 * 1000, retryPolicy);
    }

    /**
    * 建立zk连接 (Fluent风格)
    *
    * @param connString zk连接字符串, 如"192.169.7.171:2181,192.169.7.172:2181,192.169.7.173:2181"
    * @return CuratorFramework
    */
    public static CuratorFramework getConn2(String connString) {
        //重试策略 baseSleepTimeMs 重试之间等待的初始时间，maxRetries 重试的最大次数
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,5);

        return CuratorFrameworkFactory.builder()
                .connectString(connString)
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .namespace("jms")  // 根节点名称设置
                .build();
    }

3.创建节点

/**
* 创建节点
*
* @param client    客户端对象
* @param path      路径
* @param data      数据
* @param mode      创建模式
* @throws Exception
*/
public static void create(CuratorFramework client, String path, String data, CreateMode mode) {
    try {
        client.create().withMode(mode).forPath(path, data.getBytes());
    } catch (Exception e) {
        e.printStackTrace();
    }
}

4.查询节点

/**
* 查询节点数据
*
* @param client    客户端对象
* @param path      路径
*/
public static String getData(CuratorFramework client, String path) {
    try {
        return new String(client.getData().forPath(path));
    } catch (Exception e) {
        e.printStackTrace();
    }

    return "";
}

/**
* 查询子节点数据
*
* @param client    客户端对象
* @param path      路径
*/
public static List<String> getChildren(CuratorFramework client, String path) {
    try {
        return client.getChildren().forPath(path);
    } catch (Exception e) {
        e.printStackTrace();
    }

    return null;
}

/**
* 查询节点信息
*
* @param client    客户端对象
* @param path      路径
*/
public static Stat getStatus(CuratorFramework client, String path) {
    try {
        Stat status = new Stat();
        client.getData().storingStatIn(status).forPath(path);

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

    return null;
}

5.更新节点

/**
* 更新节点
*
* @param client    客户端对象
* @param path      路径
* @param data      数据
*/
public static void update(CuratorFramework client, String path, String data) {
    try {
        client.setData().forPath(path, data.getBytes());
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
* 指定版本号更新节点(如果版本号变了会抛出异常)
*
* @param client    客户端对象
* @param path      路径
* @param data      数据
*/
public static void updateWithVersion(CuratorFramework client, String path, String data) {
    try {
        // 查询节点状态信息
        Stat status = new Stat();
        client.getData().storingStatIn(status).forPath(path);
        int version = status.getVersion();

        client.setData().withVersion(version).forPath(path, data.getBytes());
    } catch (Exception e) {
        e.printStackTrace();
    }
}

6.删除节点

/**
* 删除节点
*
* @param client    客户端对象
* @param path      路径
*/
public static void delete(CuratorFramework client, String path) {
    try {
        client.delete().forPath(path);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
* 删除带有子节点的节点
*
* @param client    客户端对象
* @param path      路径
*/
public static void deletingChildrenIfNeeded(CuratorFramework client, String path) {
    try {
        client.delete().deletingChildrenIfNeeded().forPath(path);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
* 删除节点(超时重试删除和带回调功能)
*
* @param client    客户端对象
* @param path      路径
* @param callback  删除回调
*/
public static void deletingGuaranteed(CuratorFramework client, String path, BackgroundCallback callback) {
    try {
        client.delete().guaranteed().inBackground(callback).forPath(path);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

7.测试代码

public static void main(String[] args) {
    // 1.建立连接
    String connString = "192.169.7.171:2181";
    CuratorFramework client = ConnectionExample.getConn2(connString);

    client.start();

    // 2.创建节点
    create(client, "/test1", "测试1", CreateMode.PERSISTENT); // 持久节点
    create(client, "/test2", "测试2", CreateMode.PERSISTENT_SEQUENTIAL); // 持久顺序节点
    create(client, "/test3", "测试3", CreateMode.EPHEMERAL); // 临时节点
    create(client, "/test4", "测试4", CreateMode.EPHEMERAL_SEQUENTIAL); // 临时顺序节点

    // 3.查询节点
    System.out.println(getData(client, "/test1"));
    System.out.println(getChildren(client, "/"));
    System.out.println(getStatus(client, "/test1"));

    // 4.修改节点
    update(client, "/test1", "测试修改1");
    updateWithVersion(client, "/test1", "测试修改2");

    // 5.删除节点
    deletingGuaranteed(client, "/test1", new BackgroundCallback() {
        @Override
        public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
            System.out.println(event);
        }
    });

    // 6.关闭连接
    try {
        Thread.sleep(60000L);

        client.close();
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

源码

代码仓库地址: https://github.com/james-java/curator-example