Curator 使用 (一)
POM 依赖
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>4.1.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.1.0</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.13</version>
</dependency>
客户端初始化
客户端代码初始化如下所示：
CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString(zkServerPath)
.sessionTimeoutMs(10000)
.retryPolicy(retryPolicy)
.namespace("workspace")
.build();
重试策略
客户端是使用建造者模式进行对象的构造，其中非常重要的一个参数是 retryPolicy 用于表示客户端的超时重连策略。目前所提供的超时重连策略有如下几种：
- ExponentialBackoffRetry：重试指定的次数, 且每一次重试之间停顿的时间逐渐增加
/**
* 同步创建 zk 示例，原生 api 是异步的
*
* curator 链接 zookeeper 的策略:ExponentialBackoffRetry
* baseSleepTimeMs：初始 sleep 的时间
* maxRetries：最大重试次数
* maxSleepMs：最大重试时间
*/
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
- RetryNTimes：指定最大重试次数的重试策略
/**
* curator 链接 zookeeper 的策略:RetryNTimes
* n：重试的次数
* sleepMsBetweenRetries：每次重试间隔的时间
*/
RetryPolicy retryPolicy = new RetryNTimes(3, 5000);
- RetryOneTime：仅重试一次
/**
* curator 链接 zookeeper 的策略:RetryOneTime
* sleepMsBetweenRetry:每次重试间隔的时间
*/
RetryPolicy retryPolicy = new RetryOneTime(3000);
- RetryForever：永远重试
/**
* retryIntervalMs:每次重试间隔的时间
*/
RetryPolicy retryPolicy = new RetryForever(retryIntervalMs)
- RetryUntilElapsed：一直重试直到达到规定的时间
/**
* curator 链接 zookeeper 的策略:RetryUntilElapsed
* maxElapsedTimeMs:最大重试时间
* sleepMsBetweenRetries:每次重试间隔
* 重试时间超过 maxElapsedTimeMs 后，就不再重试
*/
RetryPolicy retryPolicy4 = new RetryUntilElapsed(2000, 3000);
客户端启动
client.start();
判断客户端状态
可以通过如下的代码判断当前客户端状态是否已经建立连接：
boolean isZkCuratorStarted = client.getState() == CuratorFrameworkState.STARTED;
System.out.println("当前客户的状态：" + (isZkCuratorStarted ? "连接中" : "已关闭"));
其中客户端的状态信息都存储在 CuratorFrameworkState 类中，其中包含的状态如下所示：
/**
* @see CuratorFramework#getState()
*/
public enum CuratorFrameworkState {
/**
* {@link CuratorFramework#start()} has not yet been called
*/
LATENT,
/**
* {@link CuratorFramework#start()} has been called
*/
STARTED,
/**
* {@link CuratorFramework#close()} has been called
*/
STOPPED
}
客户端常用操作
创建节点
// 创建节点
String nodePath = "/super/imooc";
byte[] data = "superme".getBytes();
client.create().creatingParentsIfNeeded()
.withMode(CreateMode.PERSISTENT)
.withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
.forPath(nodePath, data);
这里需要注意几点：
- reatingParentsIfNeeded()：如果父节点路径不存在，可以递归创建父节点
- withMode(CreateMode.PERSISTENT)：用于指定节点的类型
public enum CreateMode {
/** PERSISTENT：持久节点
* PERSISTENT_SEQUENTIAL：持久顺序节点
* EPHEMERAL：临时节点
* EPHEMERAL_SEQUENTIAL：临时顺序节点
*/
PERSISTENT, PERSISTENT_SEQUENTIAL, EPHEMERAL, EPHEMERAL_SEQUENTIAL;
}
- withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)：设置 Znode 的访问权限
public static interface Ids {
/**
* OPEN_ACL_UNSAFE --> world:anyone:cdrwa
* CREATOR_ALL_ACL --> auth:user:password:cdrwa
*/
org.apache.zookeeper.data.Id ANYONE_ID_UNSAFE;
org.apache.zookeeper.data.Id AUTH_IDS;
java.util.ArrayList<org.apache.zookeeper.data.ACL> OPEN_ACL_UNSAFE;
java.util.ArrayList<org.apache.zookeeper.data.ACL> CREATOR_ALL_ACL;
java.util.ArrayList<org.apache.zookeeper.data.ACL> READ_ACL_UNSAFE;
}
更新节点数据
byte[] newData = "batman".getBytes();
client.setData()
.withVersion(0)
.forPath(nodePath, newData);
Zookeeper 中包含乐观锁，因此在更新数据的时候需要注意节点本身的数据版本withVersion(0)。
删除节点
// 删除节点
client.delete()
.guaranteed() // 如果删除失败，那么在后端还是继续会删除，直到成功
.deletingChildrenIfNeeded() // 如果有子节点，就删除
.withVersion(0)
.forPath(nodePath);
读取节点数据
Stat stat = new Stat();
byte[] data = client.getData()
.storingStatIn(stat)
.forPath(nodePath);
System.out.println("节点" + nodePath + "的数据为: " + new String(data));
System.out.println("该节点的版本号为: " + stat.getVersion());
