Apache Curator操作zookeeper的API使用

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;// 集群模式则是多个ipprivate 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;// 集群模式则是多个ipprivate 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;// 集群模式则是多个ipprivate 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节点的增删查改操作就演示完毕了。