Zookeeper客户端curator常用法

Maven配置

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

基本用法

public static void main( String[] args ) throws Exception {
//创建重连策略
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

//创建会话
CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString("localhost:2181")
.sessionTimeoutMs(5000)
.retryPolicy(retryPolicy)
.build();

client.start();

String path = "/example";
//创建节点
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, "init".getBytes());

//读取数据节点
byte[] data = client.getData().forPath(path);
System.out.println("init data is: " + new String(data));

//更新数据节点
client.setData().forPath(path, "update data".getBytes());

//删除数据节点
client.delete().deletingChildrenIfNeeded().forPath(path);

}

异步接口

//请参考前面代码

...

//关键代码

final CountDownLatch cdl = new CountDownLatch(1);
String path = "/example";
//创建节点
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
//创建成功后回调方法
.inBackground(new BackgroundCallback() {
@Override
public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
System.out.println("node name is: " + curatorEvent.getName());
System.out.println("node path is: " + curatorEvent.getPath());
System.out.println("event type: " + curatorEvent.getType());
cdl.countDown();
}
})
.forPath(path, "init".getBytes());
System.out.println("already commit!");

cdl.await();

already commit!

node name is: /example

node path is: /example

event type: CREATE

事件监听

数据节点监听

public static void watchNode() throws Exception {
client.start();
//创建节点
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, "init".getBytes());

final NodeCache nodeCache = new NodeCache(client, path);
nodeCache.start();

final CountDownLatch countDownLatch = new CountDownLatch(1);
nodeCache.getListenable().addListener(new NodeCacheListener() {
@Override
public void nodeChanged() throws Exception {
byte[] data = nodeCache.getCurrentData().getData();
System.out.println(new String(data));
countDownLatch.countDown();
}
});

client.setData().forPath(path, "update node".getBytes());
countDownLatch.await();
}

update node

数据子节点监听

public static void watchChildrenNode() throws Exception {
client.start();
//创建节点

final NodeCache nodeCache = new NodeCache(client, path);
nodeCache.start();

final CountDownLatch countDownLatch = new CountDownLatch(2);
PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);
pathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
@Override
public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
System.out.println("event type: " + pathChildrenCacheEvent.getType());
System.out.println("node data: " + pathChildrenCacheEvent.getData());
countDownLatch.countDown();
}
});
client.create().withMode(CreateMode.EPHEMERAL).forPath(path + "/child", "child data".getBytes());
System.out.println("init child data: " + new String(client.getData().forPath(path + "/child")));

client.setData().forPath(path + "/child", "set child data".getBytes());
countDownLatch.await();
}

event type: INITIALIZED

node data: null

init child data: child data

event type: CHILD_ADDED

node data: ChildData{path='/example/child', stat=151,151,1453725266116,1453725266116,0,0,0,95263405746815019,10,0,151

, data=[99, 104, 105, 108, 100, 32, 100, 97, 116, 97]}