通过Curator操作Zookeeper的简单例子代码

Curator主要解决了三类问题: 一个是ZooKeeper client与ZooKeeper server之间的连接处理; 一个是提供了一套Fluent风格的操作API; 一个是ZooKeeper各种应用场景(recipe, 比如共享锁服务, 集群领导选举机制)抽象封装.

Curator几个组成部分

Client: 是ZooKeeper客户端的一个替代品, 提供了一些底层处理和相关的工具方法.
Framework: 用来简化ZooKeeper高级功能的使用, 并增加了一些新的功能, 比如管理到ZooKeeper集群的连接, 重试处理
Recipes: 实现了通用ZooKeeper的recipe, 该组件建立在Framework的基础之上
Utilities:各种ZooKeeper的工具类
Errors: 异常处理, 连接, 恢复等.
Extensions: recipe扩展

Framework 方法说明:

create(): 发起一个create操作. 可以组合其他方法 (比如mode 或background) 最后以forPath()方法结尾
delete(): 发起一个删除操作. 可以组合其他方法(version 或background) 最后以forPath()方法结尾
checkExists(): 发起一个检查ZNode 是否存在的操作. 可以组合其他方法(watch 或background) 最后以forPath()方法结尾
getData(): 发起一个获取ZNode数据的操作. 可以组合其他方法(watch, background 或get stat) 最后以forPath()方法结尾
setData(): 发起一个设置ZNode数据的操作. 可以组合其他方法(version 或background) 最后以forPath()方法结尾
getChildren(): 发起一个获取ZNode子节点的操作. 可以组合其他方法(watch, background 或get stat) 最后以forPath()方法结尾
inTransaction(): 发起一个ZooKeeper事务. 可以组合create, setData, check, 和/或delete 为一个操作, 然后commit() 提交

import java.awt.Event;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorEventType;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

public class TestCurator {

public static void main(String[] args) throws Exception {
RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000);
ExecutorService es = Executors.newFixedThreadPool(5); // 线程池
CuratorFramework client = CuratorFrameworkFactory.builder().connectString("192.168.1.134")
.sessionTimeoutMs(5000).connectionTimeoutMs(5000).retryPolicy(retryPolicy).build();
client.start();

// 1.创建节点 (临时节点）
// String path1 =
// client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/node_1",
// "I'm NODE_1".getBytes());
// System.out.println("创建node_1" + path1);

// String path2 =
// client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
// .forPath("/node_1/node_1_1", "I'm NODE_1_1".getBytes());
// System.out.println("创建" + path2);

// String path3 =
// client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
// .forPath("/node_1/node_1_3", "I'm NODE_1_3".getBytes());
// System.out.println("创建" + path3);

// 2.删除节点
// 普通单节点：直接指定删除路径；需要服务端进行版本检验：withVersion；
// 删除子节点：deletingChildrenIfNeeded；保障机制：guarangteed。
// client.delete().guaranteed().deletingChildrenIfNeeded().forPath("/node_1/1");

// 3.查看子节点列表
List<String> clist = client.getChildren().forPath("/node_1");
System.out.println("子节点： " + clist.toString());

// 4.获取节点的数据内容返回字节数组；获取节点状态信息：storingStatIN
Stat stat = new Stat();
byte[] ret = client.getData().storingStatIn(stat).forPath("/node_1");
System.out.println("数据: " + new String(ret));
System.out.println("节点状态信息: " + stat);

// 5.修改数据
client.setData().withVersion(stat.getVersion()).forPath("/node_1", "我是node_1".getBytes());
byte[] ret1 = client.getData().storingStatIn(stat).forPath("/node_1");
System.out.println("修改后的数据: " + new String(ret1));

// 6.查看是否存在 Stat s = client.checkExists().forPath("") 存在则返回stat对象，否则返回空
// 异步调用
client.checkExists().inBackground(new BackgroundCallback() {

public void processResult(CuratorFramework arg0, CuratorEvent arg1) throws Exception {
// TODO Auto-generated method stub
CuratorEventType t = arg1.getType();
System.out.println("事件类型:"+t);
System.out.println("返回码: " + arg1.getResultCode());// 返回码成功返回码为0
System.out.println("触发事件的节点路径: " + arg1.getPath());
System.out.println("子节点: " + arg1.getChildren());
System.out.println("数据内容: " + arg1.getData());
System.out.println("节点状态信息1: " + arg1.getStat());
System.out.println("上下文: " + arg1.getContext()); // 上下文，执行异步调用时传入额外参数供我们使用
}
}, "123", es).forPath("/node_1");

// 7.节点监听器只能监听节点的新增，及修改，不能对节点的删除进行监听处理。
final NodeCache cache = new NodeCache(client, "/node_1");
cache.start();
cache.getListenable().addListener(new NodeCacheListener() {

public void nodeChanged() throws Exception {
// TODO Auto-generated method stub
byte[] ret = cache.getCurrentData().getData(); //拿到当前结点的最新数据
System.out.println("新数据： " + new String(ret));
}
});

// 8。子节点监听器子节点的添加，修改，删除
final PathChildrenCache cache2 = new PathChildrenCache(client, "/node_1", true);
cache2.start();
cache2.getListenable().addListener(new PathChildrenCacheListener() {

public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
// TODO Auto-generated method stub
switch (event.getType()) {
case CHILD_ADDED:
System.out.println("CHILD_ADDED:"+event.getData());
break;
case CHILD_UPDATED:
System.out.println("CHILD_UPDATED:"+event.getData());
break;
case CHILD_REMOVED:
System.out.println("CHILD_REMOVED:"+event.getData());
break;
default:
break;
}

}
});

Thread.sleep(Integer.MAX_VALUE);

}
}