说明
注意: 如未安装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);
}
}
