Curator 是操作ZooKeeper的开源客户端.
类似于JDBC什么的吧.
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.4.2</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>2.4.2</version>
</dependency>

同步接口
1.创建节点

public static void main(String[] args) throws Exception {
CuratorFramework client = getConnection();
client.create().forPath("/t1", "t1".getBytes());
client.create().withMode(CreateMode.EPHEMERAL).forPath("/e1","e1".getBytes());
client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/p1/p2/p3","p1".getBytes());
client.close();
}
public static CuratorFramework getConnection() {
CuratorFramework client;
client = CuratorFrameworkFactory.newClient("172.16.1.78:2181", new ExponentialBackoffRetry(1000, 3));
client.start();
return client;
}

节点类型一共有四种
CreateMode.EPHEMERAL 临时节点
CreateMode.EPHEMERAL_SEQUENTIAL 临时顺序节点
CreateMode.PERSISTENT 持久节点
CreateMode.PERSISTENT_SEQUENTIAL 持久顺序节点

2.删除节点

public static void main(String[] args) throws Exception {
CuratorFramework client = getConnection();
//递归删除子节点
client.delete().deletingChildrenIfNeeded().forPath("/p1");
//强制保证删除
client.delete().guaranteed().forPath("/t1");
client.close();
}

3.读取节点数据

public static void main(String[] args) throws Exception {
CuratorFramework client = getConnection();
byte[] data=client.getData().forPath("/t");
System.out.println(new String(data));
}

4.更新节点数据

public static void main(String[] args) throws Exception {
CuratorFramework client = getConnection();
client.setData().forPath("/t","Hello World".getBytes());
client.close();
}

异步接口

1.异步接口调用BackgroundCallback回调函数

public static void main(String[] args) throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
ExecutorService threadpool = Executors.newFixedThreadPool(5);
BackgroundCallback callback = new BackgroundCallback() {
public void processResult(CuratorFramework arg0, CuratorEvent event) throws Exception {
System.out.println(event.getType().name());
System.out.println("EventCode:" + event.getResultCode());
System.out.println(Thread.currentThread().getName());
latch.countDown();
}
};
CuratorFramework client = getConnection();
client.delete().inBackground(callback, threadpool).forPath("/t");
latch.await();
threadpool.shutdown();
client.close();
}

结果:
DELETE
EventCode:-101
pool-1-thread-1

2.事件监听

监听节点数据变化.
监听的节点可以不存在,但是一旦创建也会触发事件.
删除监听的节点,不会触发事件.

public static void main(String[] args) throws Exception {
CuratorFramework client = getConnection();
final NodeCache cache = new NodeCache(client, "/t");
cache.start(true);
cache.getListenable().addListener(new NodeCacheListener() {
public void nodeChanged() throws Exception {
byte[] data = cache.getCurrentData().getData();
System.out.println(new String(data));
}
});
Thread.sleep(Integer.MAX_VALUE);
}

监听子节点变化

public static void main(String[] args) throws Exception {
CuratorFramework client = getConnection();
final PathChildrenCache cache = new PathChildrenCache(client, "/t", true);
cache.start(StartMode.POST_INITIALIZED_EVENT);
cache.getListenable().addListener(new PathChildrenCacheListener() {
public void childEvent(CuratorFramework arg0, PathChildrenCacheEvent event) throws Exception {
System.out.println(event.getType().name());
System.out.println(event.getData().getPath());
}
});
Thread.sleep(Integer.MAX_VALUE);
}

子节点变化包括
event.getType()
CHILD_ADDED 子节点创建
CHILD_UPDATED 子节点数据变更
CHILD_REMOVED 子节点删除

3.Master选举

public static void main(String[] args) throws Exception {
CuratorFramework client = getConnection();
LeaderSelector selector=new LeaderSelector(client,"/master",new LeaderSelectorListenerAdapter() {
public void takeLeadership(CuratorFramework client) throws Exception {
System.out.println("成为Master角色");
Thread.sleep(1000);
System.out.println("释放");
}
});
selector.autoRequeue();
selector.start();
Thread.sleep(Integer.MAX_VALUE);
}

4.分布式锁

public static void main(String[] args) throws Exception {
CuratorFramework client = getConnection();
final InterProcessMutex lock = new InterProcessMutex(client, "/lock");
final CountDownLatch latch = new CountDownLatch(1);
for (int i = 0; i < 30; i++) {
new Thread(new Runnable() {
public void run() {
try {
latch.await();
lock.acquire();
System.out.println(new Date());
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
lock.release();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}).start();
}
latch.countDown();
}

5.分布式计数器

public static void main(String[] args) throws Exception {
CuratorFramework client = getConnection();
DistributedAtomicInteger atomicInteger = new DistributedAtomicInteger(client, "/atomicInteger",
new RetryNTimes(3, 1000));
AtomicValue<Integer> rc = atomicInteger.add(8);
System.out.println(rc.succeeded());
System.out.println(rc.postValue());
}

6.分布式Barrier

public static void main(String[] args) throws Exception {
final CountDownLatch latch = new CountDownLatch(30);
for (int i = 0; i < 30; i++) {
new Thread(new Runnable() {
public void run() {
try {
CuratorFramework client = getConnection();
DistributedBarrier barrier = new DistributedBarrier(client, "/barrier");
barrier.setBarrier();
System.out.println(Thread.currentThread().getName() + ",waiting..");
latch.countDown();
barrier.waitOnBarrier();
System.out.println(Thread.currentThread().getName() + ",start!!");
client.close();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}).start();
}
latch.await();
CuratorFramework client = getConnection();
DistributedBarrier barrier = new DistributedBarrier(client, "/barrier");
barrier.removeBarrier();
client.close();
}

参考:
http://ifeve.com/zookeeper-path-cache/

ITPub博客

Curator API使用

最新文章