Apache Curator操作zookeeper的API使用

旧城等待， 2021-06-24 16:09 445阅读 0赞

### curator简介与客户端之间的异同点 ###

常用的zookeeper java客户端：

*  zookeeper原生Java API
 *  zkclient
 *  Apache curator

ZooKeeper原生Java API的不足之处：

*  在连接zk超时的时候，不支持自动重连，需要手动操作
 *  Watch注册一次就会失效，需要反复注册
 *  不支持递归创建节点

Apache curator：

*  Apache 的开源项目
 *  解决Watch注册一次就会失效的问题
 *  提供的 API 更加简单易用
 *  提供更多解决方案并且实现简单，例如：分布式锁
 *  提供常用的ZooKeeper工具类
 *  编程风格更舒服

创建一个普通的maven工程，在pom.xml文件中，配置如下依赖：

<dependencies>
            <dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>3.4.11</version>
            </dependency>
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-framework</artifactId>
                <version>4.0.1</version>
            </dependency>
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-recipes</artifactId>
                <version>4.0.1</version>
            </dependency>
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-lang3</artifactId>
                <version>3.5</version>
            </dependency>
            <dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-databind</artifactId>
                <version>2.7.4</version>
            </dependency>
            <dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-core</artifactId>
                <version>2.7.4</version>
            </dependency>
            <dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-annotations</artifactId>
                <version>2.7.4</version>
            </dependency>
        </dependencies>

配置完依赖后，我们就可以来写一个简单的demo测试与zookeeper服务端的连接。代码如下：

package org.zero01.zk.curator;
     
    import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.retry.ExponentialBackoffRetry;
     
    /**
     * @program: zookeeper-connection
     * @description: 建立curator与zkserver的连接演示demo
     * @author: 01
     * @create: 2018-04-28 09:44
     **/
    public class CuratorConnect {
     
        // Curator客户端
        public CuratorFramework client = null;
        // 集群模式则是多个ip
        private static final String zkServerIps = "192.168.190.128:2181,192.168.190.129:2181,192.168.190.130:2181";
     
        public CuratorConnect(){
            /**
             * 同步创建zk示例，原生api是异步的
             * 这一步是设置重连策略
             *
             * ExponentialBackoffRetry构造器参数：
             *  curator链接zookeeper的策略:ExponentialBackoffRetry
             *  baseSleepTimeMs：初始sleep的时间
             *  maxRetries：最大重试次数
             *  maxSleepMs：最大重试时间
             */
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
     
            // 实例化Curator客户端，Curator的编程风格可以让我们使用方法链的形式完成客户端的实例化
            client = CuratorFrameworkFactory.builder() // 使用工厂类来建造客户端的实例对象
                    .connectString(zkServerIps)  // 放入zookeeper服务器ip
                    .sessionTimeoutMs(10000).retryPolicy(retryPolicy)  // 设定会话时间以及重连策略
                    .build();  // 建立连接通道
     
            // 启动Curator客户端
            client.start();
        }
     
        // 关闭zk客户端连接
        private void closeZKClient() {
            if (client != null) {
                this.client.close();
            }
        }
     
        public static void main(String[] args) throws InterruptedException {
            // 实例化
            CuratorConnect curatorConnect = new CuratorConnect();
            // 获取当前客户端的状态
            boolean isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
     
            Thread.sleep(1000);
     
            // 关闭客户端
            curatorConnect.closeZKClient();
     
            // 获取当前客户端的状态
            isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
        }
    }

输出

当前客户端的状态：连接中...
    当前客户端的状态：已关闭...

curator连接zookeeper服务器时有自动重连机制，而curator的重连策略有五种。第一种就是我们以上demo中使用到的：

/**
     * （推荐）
     * 同步创建zk示例，原生api是异步的
     * 这一步是设置重连策略
     * 
     * 构造器参数：
     *  curator链接zookeeper的策略:ExponentialBackoffRetry
     *  baseSleepTimeMs：初始sleep的时间
     *  maxRetries：最大重试次数
     *  maxSleepMs：最大重试时间
     */
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);

第二种，可设定重连n次：

/**
     * （推荐）
     * curator链接zookeeper的策略:RetryNTimes
     * 
     * 构造器参数：
     * n：重试的次数
     * sleepMsBetweenRetries：每次重试间隔的时间
     */
    RetryPolicy retryPolicy = new RetryNTimes(3, 5000);

第三种，只会重连一次：

/**
     * （不推荐）
     * curator链接zookeeper的策略:RetryOneTime
     * 
     * 构造器参数：
     * sleepMsBetweenRetry:每次重试间隔的时间
     * 这个策略只会重试一次
     */
    RetryPolicy retryPolicy2 = new RetryOneTime(3000);

第四种，永远重连：

/**
     * 永远重试，不推荐使用
     */
    RetryPolicy retryPolicy3 = new RetryForever(retryIntervalMs)

第五种，可设定最大重试时间：

/**
     * curator链接zookeeper的策略:RetryUntilElapsed
     * 
     * 构造器参数：
     * maxElapsedTimeMs:最大重试时间
     * sleepMsBetweenRetries:每次重试间隔
     * 重试时间超过maxElapsedTimeMs后，就不再重试
     */
    RetryPolicy retryPolicy4 = new RetryUntilElapsed(2000, 3000);

### zk命名空间以及创建节点 ###

zookeeper的命名空间就类似于我们平时使用Eclipse等开发工具的工作空间一样，我们该连接中所有的操作都是基于这个命名空间的。curator提供了设置命名空间的方法，这样我们任何的连接都可以去设置一个命名空间。设置了命名空间并成功连接后，zookeeper的根节点会多出一个以命名空间名称所命名的节点。然后我们在该连接的增删查改等操作都会在这个节点中进行。例如，现在zookeeper服务器上只有以下几个节点：

[zk: localhost:2181(CONNECTED) 0] ls /
    [zookeeper, data, real-culster, testDigestNode]
    [zk: localhost:2181(CONNECTED) 1]

然后我们来将之前的demo修改一下，加上设置命名空间的代码以及创建节点的代码，以此来做一个简单的演示，修改之前的demo代码如下：

package org.zero01.zk.curator;
     
    import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.ZooDefs;
     
    public class CuratorConnect {
     
        // Curator客户端
        public CuratorFramework client = null;
        // 集群模式则是多个ip
        private static final String zkServerIps = "192.168.190.128:2181,192.168.190.129:2181,192.168.190.130:2181";
     
        public CuratorConnect() {
     
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
     
            // 实例化Curator客户端
            client = CuratorFrameworkFactory.builder() // 使用工厂类来建造客户端的实例对象
                    .connectString(zkServerIps)  // 放入zookeeper服务器ip
                    .sessionTimeoutMs(10000).retryPolicy(retryPolicy)  // 设定会话时间以及重连策略
                    .namespace("workspace").build();  // 设置命名空间以及开始建立连接
     
            // 启动Curator客户端
            client.start();
        }
     
        // 关闭zk客户端连接
        private void closeZKClient() {
            if (client != null) {
                this.client.close();
            }
        }
     
        public static void main(String[] args) throws Exception {
            // 实例化
            CuratorConnect curatorConnect = new CuratorConnect();
            // 获取当前客户端的状态
            boolean isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
     
            // 创建节点
            String nodePath = "/super/testNode";  // 节点路径
            byte[] data = "this is a test data".getBytes();  // 节点数据
            String result = curatorConnect.client.create().creatingParentsIfNeeded()  // 创建父节点，也就是会递归创建
                    .withMode(CreateMode.PERSISTENT)  // 节点类型
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)  // 节点的acl权限
                    .forPath(nodePath, data);
     
            System.out.println(result + "节点，创建成功...");
     
            Thread.sleep(1000);
     
            // 关闭客户端
            curatorConnect.closeZKClient();
     
            // 获取当前客户端的状态
            isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
        }
    }

控制台输出

当前客户端的状态：连接中...
    /super/testNode节点，创建成功...
    当前客户端的状态：已关闭...

到服务器上，查看是否多了一个 workspace 节点，并且我们创建的节点都在这个节点下：

[zk: localhost:2181(CONNECTED) 12] ls /
    [workspace, zookeeper, data, real-culster, testDigestNode]
    [zk: localhost:2181(CONNECTED) 13] ls /workspace
    [super]
    [zk: localhost:2181(CONNECTED) 14] ls /workspace/super
    [testNode]
    [zk: localhost:2181(CONNECTED) 15] get /workspace/super/testNode
    this is a test data
    cZxid = 0xb0000000f
    ctime = Sat Apr 28 18:56:36 CST 2018
    mZxid = 0xb0000000f
    mtime = Sat Apr 28 18:56:36 CST 2018
    pZxid = 0xb0000000f
    cversion = 0
    dataVersion = 0
    aclVersion = 0
    ephemeralOwner = 0x0
    dataLength = 19
    numChildren = 0
    [zk: localhost:2181(CONNECTED) 18] getAcl /workspace/super/testNode
    'world,'anyone
    : cdrwa
    [zk: localhost:2181(CONNECTED) 19]

### 修改节点以及删除节点 ###

上一节中，我们介绍了如何创建节点，本节简单演示如何修改节点的数据以及删除节点。修改 CuratorConnect 类代码如下：

package org.zero01.zk.curator;
     
    import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.ZooDefs;
    import org.apache.zookeeper.data.Stat;
     
    public class CuratorConnect {
     
        // Curator客户端
        public CuratorFramework client = null;
        // 集群模式则是多个ip
        private static final String zkServerIps = "192.168.190.128:2181,192.168.190.129:2181,192.168.190.130:2181";
     
        public CuratorConnect() {
     
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
     
            // 实例化Curator客户端
            client = CuratorFrameworkFactory.builder() // 使用工厂类来建造客户端的实例对象
                    .connectString(zkServerIps)  // 放入zookeeper服务器ip
                    .sessionTimeoutMs(10000).retryPolicy(retryPolicy)  // 设定会话时间以及重连策略
                    .namespace("workspace").build();  // 设置命名空间以及开始建立连接
     
            // 启动Curator客户端
            client.start();
        }
     
        // 关闭zk客户端连接
        private void closeZKClient() {
            if (client != null) {
                this.client.close();
            }
        }
     
        public static void main(String[] args) throws Exception {
            // 实例化
            CuratorConnect curatorConnect = new CuratorConnect();
            // 获取当前客户端的状态
            boolean isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
     
            // 节点路径
            String nodePath = "/super/testNode";
     
            // 更新节点数据
            byte[] newData = "this is a new data".getBytes();
            Stat resultStat = curatorConnect.client.setData().withVersion(0)  // 指定数据版本
                    .forPath(nodePath, newData);  // 需要修改的节点路径以及新数据
     
            System.out.println("更新节点数据成功，新的数据版本为：" + resultStat.getVersion());
     
            // 删除节点
            curatorConnect.client.delete()
                    .guaranteed()  // 如果删除失败，那么在后端还是会继续删除，直到成功
                    .deletingChildrenIfNeeded()  // 子节点也一并删除，也就是会递归删除
                    .withVersion(resultStat.getVersion())
                    .forPath(nodePath);
     
            Thread.sleep(1000);
     
            // 关闭客户端
            curatorConnect.closeZKClient();
     
            // 获取当前客户端的状态
            isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
        }
    }

输出

当前客户端的状态：连接中...
    更新节点数据成功，新的数据版本为：1
    当前客户端的状态：已关闭...

此时到zookeeper服务器上，可以看到，节点已经被成功删除了：

[zk: localhost:2181(CONNECTED) 19] ls /workspace/super
    []
    [zk: localhost:2181(CONNECTED) 20]

### 查询节点相关信息 ###

1.获取某个节点的数据，现有一个节点的数据如下：

[zk: localhost:2181(CONNECTED) 22] get /workspace/super/testNode    
    test-data
    cZxid = 0xb00000015
    ctime = Sat Apr 28 20:59:57 CST 2018
    mZxid = 0xb00000015
    mtime = Sat Apr 28 20:59:57 CST 2018
    pZxid = 0xb00000015
    cversion = 0
    dataVersion = 0
    aclVersion = 0
    ephemeralOwner = 0x0
    dataLength = 9
    numChildren = 0
    [zk: localhost:2181(CONNECTED) 23]

修改 CuratorConnect 类中的main方法代码如下，一些重复的代码就忽略了：

...
    public class CuratorConnect {
        ...
        public static void main(String[] args) throws Exception {
            // 实例化
            CuratorConnect curatorConnect = new CuratorConnect();
            // 获取当前客户端的状态
            boolean isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
     
            // 节点路径
            String nodePath = "/super/testNode";
     
            // 读取节点数据
            Stat stat = new Stat();
            byte[] nodeData = curatorConnect.client.getData().storingStatIn(stat).forPath(nodePath);
            System.out.println("节点 " + nodePath + " 的数据为：" + new String(nodeData));
            System.out.println("该节点的数据版本号为：" + stat.getVersion());
     
            Thread.sleep(1000);
     
            // 关闭客户端
            curatorConnect.closeZKClient();
     
            // 获取当前客户端的状态
            isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
        }
    }

运行该类，控制台输出内容如下：

当前客户端的状态：连接中...
    节点 /super/testNode 的数据为：test-data
    该节点的数据版本号为：0
    当前客户端的状态：已关闭...

2.获取某个节点下的子节点列表，现有一个节点的子节点列表如下：

[zk: localhost:2181(CONNECTED) 33] ls /workspace/super/testNode
    [threeNode, twoNode, oneNode]
    [zk: localhost:2181(CONNECTED) 34]

修改 CuratorConnect 类中的main方法代码如下，一些重复的代码就忽略了：

...
    public class CuratorConnect {
        ...
        public static void main(String[] args) throws Exception {
            // 实例化
            CuratorConnect curatorConnect = new CuratorConnect();
            // 获取当前客户端的状态
            boolean isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
     
            // 节点路径
            String nodePath = "/super/testNode";
     
            // 获取子节点列表
            List<String> childNodes = curatorConnect.client.getChildren().forPath(nodePath);
            System.out.println(nodePath + " 节点下的子节点列表：");
            for (String childNode : childNodes) {
                System.out.println(childNode);
            }
     
            Thread.sleep(1000);
     
            // 关闭客户端
            curatorConnect.closeZKClient();
     
            // 获取当前客户端的状态
            isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
        }
    }

输出

当前客户端的状态：连接中...
    /super/testNode 节点下的子节点列表：
    threeNode
    twoNode
    oneNode
    当前客户端的状态：已关闭...

3.查询某个节点是否存在，修改 CuratorConnect 类中的main方法代码如下，一些重复的代码就忽略了：

...
    public class CuratorConnect {
        ...
        public static void main(String[] args) throws Exception {
            // 实例化
            CuratorConnect curatorConnect = new CuratorConnect();
            // 获取当前客户端的状态
            boolean isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
     
            // 节点路径
            String nodePath = "/super/testNode";
     
            // 查询某个节点是否存在，存在就会返回该节点的状态信息，如果不存在的话则返回空
            Stat statExist = curatorConnect.client.checkExists().forPath(nodePath);
            if (statExist == null) {
                System.out.println(nodePath + " 节点不存在");
            } else {
                System.out.println(nodePath + " 节点存在");
            }
     
            Thread.sleep(1000);
     
            // 关闭客户端
            curatorConnect.closeZKClient();
     
            // 获取当前客户端的状态
            isZkCuratorStarted = curatorConnect.client.isStarted();
            System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
        }
    }

输出

当前客户端的状态：连接中...
    /super/testNode 节点存在
    当前客户端的状态：已关闭...

如果查询一个不存在的节点，会返回null，我们测试一下将nodePath改成一个不存在的节点。运行该类，输出内容如下：

当前客户端的状态：连接中...
    /super/asdasdasd 节点不存在
    当前客户端的状态：已关闭...

至此，使用curator对zookeeper节点的增删查改操作就演示完毕了。