zookeeper学习之Curator客户端

博客分类：

工具使用

Curator框架是最好用，最流行的zookeeper的客户端。

它有以下三个优点

1.提供了一套非常友好的操作API;

2. 提供一些高级特性（包括但不仅限于前篇文章中提到的）的封装

3.易测试

maven依赖如下

Xml代码

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

按照官方给出的文档和包结构，可以轻松的看出Curator功能分两大类，一是对zookeeper的一些基本命令的封装，比如增删改查。是他的framework模块，一个是他的高级特性，即recipes模块。

一、framework模块

Curator提供了一套Fluent风格的操作API。这在很多脚本类语言里比较流行。

比如他创建client的代码是这样

Java代码

CuratorFramework client = builder.connectString("192.168.11.56:2180")
.sessionTimeoutMs(30000)
.connectionTimeoutMs(30000)
.canBeReadOnly(false)
.retryPolicy(new ExponentialBackoffRetry(1000, Integer.MAX_VALUE))
.namespace(namespace)
.defaultData(null)
.build();
client.start();

一路点到底，这就是所谓的Fluent风格。

我们再看增删改查的

Java代码

public class CrudExamples {
private static CuratorFramework client = ClientFactory.newClient();
private static final String PATH = "/crud";
public static void main(String[] args) {
try {
client.start();
client.create().forPath(PATH, "I love messi".getBytes());
byte[] bs = client.getData().forPath(PATH);
System.out.println("新建的节点，data为:" + new String(bs));
client.setData().forPath(PATH, "I love football".getBytes());
// 由于是在background模式下获取的data，此时的bs可能为null
byte[] bs2 = client.getData().watched().inBackground().forPath(PATH);
System.out.println("修改后的data为" + new String(bs2 != null ? bs2 : new byte[0]));
client.delete().forPath(PATH);
Stat stat = client.checkExists().forPath(PATH);
// Stat就是对zonde所有属性的一个映射， stat=null表示节点不存在！
System.out.println(stat);
} catch (Exception e) {
e.printStackTrace();
} finally {
CloseableUtils.closeQuietly(client);
}
}
}

常用接口有

create()增

delete(): 删

checkExists(): 判断是否存在

setData(): 改

getData(): 查

所有这些方法都以forpath()结尾，辅以watch(监听)，withMode（指定模式），和inBackground（后台运行）等方法来使用。

此外，Curator还支持事务，一组crud操作同生同灭。代码如下

Java代码

/**
* 事务操作
*
* @author shencl
*/
public class TransactionExamples {
private static CuratorFramework client = ClientFactory.newClient();
public static void main(String[] args) {
try {
client.start();
// 开启事务
CuratorTransaction transaction = client.inTransaction();
Collection<CuratorTransactionResult> results = transaction.create()
.forPath("/a/path", "some data".getBytes()).and().setData()
.forPath("/another/path", "other data".getBytes()).and().delete().forPath("/yet/another/path")
.and().commit();
for (CuratorTransactionResult result : results) {
System.out.println(result.getForPath() + " - " + result.getType());
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// 释放客户端连接
CloseableUtils.closeQuietly(client);
}
}
}

这段的代码的运行结果，由于最后一步delete的节点不存在，所以整个事务commit失败。失败的原因会放在Collection<CuratorTransactionResult>中，非常友好。

好了framework部分的内容就这么多，是不是特别简单呢。下面就来看看recipes包的内容吧。。

Recipes部分提供的功能官网列的很详细，点击这里。注意文章第一段：Curator宣称，Recipes模块实现了除二阶段提交之外的所有zookeeper特性。

二、Recipes模块

主要有

Elections(选举)，Locks（锁），Barriers（关卡），Atomic（原子量），Caches，Queues等

1、 Elections

选举主要依赖于LeaderSelector和LeaderLatch2个类。前者是所有存活的客户端不间断的轮流做Leader，大同社会。后者是一旦选举出Leader，除非有客户端挂掉重新触发选举，否则不会交出领导权。某党？

这两者在实现上是可以切换的，直接上代码，怎么切换注释里有。由于篇幅所限，这里仅贴出基于LeaderSelector的选举，更多代码见附件

Java代码

/**
* 本类基于leaderSelector实现,所有存活的client会公平的轮流做leader
* 如果不想频繁的变化Leader，需要在takeLeadership方法里阻塞leader的变更！或者使用 {@link}
* LeaderLatchClient
*/
public class LeaderSelectorClient extends LeaderSelectorListenerAdapter implements Closeable {
private final String name;
private final LeaderSelector leaderSelector;
private final String PATH = "/leaderselector";
public LeaderSelectorClient(CuratorFramework client, String name) {
this.name = name;
leaderSelector = new LeaderSelector(client, PATH, this);
leaderSelector.autoRequeue();
}
public void start() throws IOException {
leaderSelector.start();
}
@Override
public void close() throws IOException {
leaderSelector.close();
}
/**
* client成为leader后，会调用此方法
*/
@Override
public void takeLeadership(CuratorFramework client) throws Exception {
int waitSeconds = (int) (5 * Math.random()) + 1;
System.out.println(name + "是当前的leader");
try {
Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
System.out.println(name + " 让出领导权\n");
}
}

Java代码

/**
* leader选举
*
* @author shencl
*/
public class LeaderSelectorExample {
public static void main(String[] args) {
List<CuratorFramework> clients = Lists.newArrayList();
List<LeaderSelectorClient> examples = Lists.newArrayList();
try {
for (int i = 0; i < 10; i++) {
CuratorFramework client = ClientFactory.newClient();
LeaderSelectorClient example = new LeaderSelectorClient(client, "Client #" + i);
clients.add(client);
examples.add(example);
client.start();
example.start();
}
System.out.println("----------先观察一会选举的结果-----------");
Thread.sleep(10000);
System.out.println("----------关闭前5个客户端，再观察选举的结果-----------");
for (int i = 0; i < 5; i++) {
clients.get(i).close();
}
// 这里有个小技巧，让main程序一直监听控制台输入，异步的代码就可以一直在执行。不同于while(ture)的是，按回车或esc可退出
new BufferedReader(new InputStreamReader(System.in)).readLine();
} catch (Exception e) {
e.printStackTrace();
} finally {
for (LeaderSelectorClient exampleClient : examples) {
CloseableUtils.closeQuietly(exampleClient);
}
for (CuratorFramework client : clients) {
CloseableUtils.closeQuietly(client);
}
}
}
}

2、locks

curator lock相关的实现在recipes.locks包里。顶级接口都是InterProcessLock。我们直接看最有代表性的 InterProcessReadWriteLock 进程内部读写锁（可重入读写锁）。什么叫可重入，什么叫读写锁。不清楚的先查好资料吧。总之读写锁一定是成对出现的。简易传送门

我们先定义两个任务，可并行的执行的，和互斥执行的。

Java代码