Zookeeper ZkClient简介

直接使用zk的api实现业务功能比较繁琐。因为要处理session loss，session expire等异常，在发生这些异常后进行重连。又因为ZK的watcher是一次性的，如果要基于wather实现发布/订阅模式，还要自己包装一下，将一次性订阅包装成持久订阅。另外如果要使用抽象级别更高的功能，比如分布式锁，leader选举等，还要自己额外做很多事情。这里介绍下ZK的两个第三方客户端包装小工具，可以分别解决上述小问题。

先看看zookeeper本身自带的客户端的问题。

1) ZooKeeper的Watcher是一次性的，用过了需要再注册；

2) session的超时后没有自动重连，生产环境中如果网络出现不稳定情况，那么这种情况出现的更加明显；

3) 没有领导选举机制，集群情况下可能需要实现stand by，一个服务挂了，另一个需要接替的效果；

4) 客户端只提供了存储byte数组的接口，而项目中一般都会使用对象。

5) 客户端接口需要处理的异常太多，并且通常，我们也不知道如何处理这些异常。

zkClient

zkClient主要做了两件事情。一件是在session loss和session expire时自动创建新的ZooKeeper实例进行重连。另一件是将一次性watcher包装为持久watcher。后者的具体做法是简单的在watcher回调中，重新读取数据的同时再注册相同的watcher实例。

I0Itec这个zookeeper客户端基本上解决了上面的所有问题，主要有以下特性：

1) 提供了zookeeper重连的特性------能够在断链的时候,重新建立连接,无论session失效与否.

2) 持久的event监听器机制------ ZKClient框架将事件重新定义分为了stateChanged、znodeChanged、dataChanged三种情况，用户可以注册这三种情况下的监听器（znodeChanged和dataChanged和路径有关），而不是注册Watcher。

3) zookeeper异常处理-------zookeeper中繁多的Exception,以及每个Exception所需要关注的事情各有不同，I0Itec简单的做了封装。

4) data序列化------简单的data序列化.(Serialzer/Deserialzer)

5）有默认的领导选举机制

请注意使用I0Itect-zkClient暂时有几个方法仍需要重写:

1) create方法 : 创建节点时,如果节点已经存在,仍然抛出NodeExistException,可是我期望它不在抛出此异常。

2) retryUtilConnected : 如果向zookeeper请求数据时(create,delete,setData等),此时链接不可用,那么调用者将会被阻塞直到链接建立成功;不过我仍然需要一些方法是非阻塞的,如果链接不可用,则抛出异常,或者直接返回。

3) create方法 : 创建节点时,如果节点的父节点不存在,我期望同时也要创建父节点,而不是抛出异常。

4) data监测: 我需要提供一个额外的功能来补充watch的不足,开启一个线程,间歇性的去zk server获取指定的path的data,并缓存起来。归因与watch可能丢失,以及它不能持续的反应znode数据的每一次变化,所以只能手动去同步获取。

zkClient简单的使用样例如下：

下面是订阅children变化

public static void zkChildChange(final String serverList) {
ZkClient zc = new ZkClient(serverList);
zc.subscribeChildChanges(PATH, new IZkChildListener() {
@Override
public void handleChildChange(String parentPath, List currentChilds) throws Exception {
System.out.println("clildren of path " + parentPath + ":" + currentChilds);
}
});
}

下面是订阅数据变化

public static void zkDataChange(final String serverList){
ZkClient zc = new ZkClient(serverList);
zc.subscribeDataChanges(PATH, new IZkDataListener() {
@Override
public void handleDataChange(String dataPath, Object data) throws Exception {
System.out.println("Data of " + dataPath + " has changed");
}

@Override
public void handleDataDeleted(String dataPath) throws Exception {
System.out.println(dataPath + " has deleted");
}
});
}

下面是订阅连接状态的变化

public static void zkStateChange(final String serverList){
ZkClient zc = new ZkClient(serverList);
zc.subscribeStateChanges(new IZkStateListener() {
@Override
public void handleNewSession() throws Exception {
System.out.println("handleNewSession()");
}

@Override
public void handleStateChanged(KeeperState stat) throws Exception {
System.out.println("handleStateChanged,stat:" + stat);
}

@Override
public void handleSessionEstablishmentError(Throwable error)
throws Exception {
System.out.println("handleSessionEstablishmentError,stat:" + error.getMessage());

}
});
}

项目中使用maven引用

<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.8</version>
</dependency>

如果发生session close、authFail和invalid，那么所有类型的wather都会被触发。

zkClient除了做了一些便捷包装之外，对watcher使用做了一点增强。比如subscribeChildChanges实际上是通过exists和getChildren关注了两个事件。这样当create(“/path”)时，对应path上通过getChildren注册的listener也会被调用。另外subscribeDataChanges实际上只是通过exists注册了事件。因为从上表可以看到，对于一个更新，通过exists和getData注册的watcher要么都会触发，要么都不会触发。

getData,getChildren(),exists()这三个方法可以针对参数中的path设置watcher，当path对应的Node 有相应变化时，server端会给对应的设置了watcher的client 发送一个一次性的触发通知事件。客户端在收到这个触发通知事件后，可以根据自己的业务逻辑进行相应地处理。

注意这个watcher的功能是一次性的，如果还想继续得到watcher通知，在处理完事件后，要重新register。

参考资料
http://www.cnblogs.com/viviman/archive/2013/03/11/2954118.html zookeeper如何永久监听
http://www.cnblogs.com/yql1986/p/4116483.html 调用zkclient (maven 地址见下面)实现监听当某个节点的数据发生变化时，将变化的信息打印到控制台。