Zookeeper 集群管理应用

使用场景

整个系统中分master和server两种角色，每个master、server都需要知道集群内当前可用的server列表。

原处理方式：master负责集群server信息的收集和分发。server启动后和master建立长连接并心跳检测，master收集server的新增和disconnect，并将收集到的信息分发给所有server。

原存在问题：master和server耦合较重；master比较难扩展。

使用方式

概述

每个server对应一个Zookeeper的一个子节点，所有master和push通过获取所有子节点取得server列表，当server disconnect时Zookeeper通知所有机器重新获取Server列表。

详细

根节点路劲：/service/serverlist

1.初始化Zookeeper连接

publicstatic void initZk() {
try{
#####初始化，如果当前有alive的zk连接就不初始化了
if(zk == null|| !zk.getState().isAlive()) {
synchronized(obj) {
if(zk == null|| !zk.getState().isAlive()) {
###把之前的close掉
closeZk();
PropertyReader p = newPropertyReader(ZK_PROPERTIES_FILE);
String zkServerStr = p.getValueAsString("zookeeper.server", ZK_SERVER_STR_DEFAULT);
intsessionTimeout = p.getValueAsInt("zookeeper.sessionTimeout", ZK_SESSION_TIMEOUT);
###重新建立连接
zk = newZooKeeper(zkServerStr, sessionTimeout,newDefaultWatcher());
}
}
}
}catch(Exception e) {
thrownew RuntimeException("Zookeeper Init Error!", e);
}
}

2.server在启动时添加代表自己的临时节点并watch其父节点

####如果之前myNodeName已经存在，则删除
ZkUtils.deleteNode(myNodeName);
####先看看父节点有没有，没有就创建下，父节点得是持久的
ZkUtils.createNode(rootNode,CreateMode.PERSISTENT);
####添加对父节点child变动的watcher
ZkUtils.addChildWatch(rootNode,ServerListWatcher.getInstance());
###创建自己的节点mynode
ZkUtils.createNode(myNodeName,CreateMode.EPHEMERAL);

3.master在启动时watch父节点

####先看看父节点有没有，没有就创建下，父节点得是持久的
ZkUtils.createNode(rootNode,CreateMode.PERSISTENT);
####添加对父节点child变动的watcher
ZkUtils.addChildWatch(rootNode,ServerListWatcher.getInstance());

4.监听该父节点的child变化并处理

publicvoid process(WatchedEvent event) {
try{
###如果是expired，需要重新初始化一遍
if(event.getState() == KeeperState.Expired) {
while(true) {
try{
// 重新初始化zk连接
ZkUtils.initZk();
// 重新初始化相关zk节点
PushZkUtils.initPushZkNodes();
// 获取nodeList
getAndSaveNodeList();
// 如果成功，则break;
break;
}catch(RuntimeException e) {
AsyncLogger.getInstance().info(LogManager.SERVICE_LOG,"ServerListWatcher:Init Error!");
// 5s后重新执行
Thread.sleep(5* 1000);
}
}
}
###如果是断开重连，需要重新获取下list，以免丢失事件
if(event.getState() == KeeperState.SyncConnected) {
getAndSaveNodeList();
}
###如果是child变化，及PushServer列表有变，则获取列表，保存本地
if(event.getType() == EventType.NodeChildrenChanged) {
AsyncLogger.getInstance().info(LogManager.SERVICE_LOG,"ServerListWatcher:Node Child Changed Event Received");
// 保存到本地，等待同步线程检查
getAndSaveNodeList();
}
}catch(Exception e) {
}
}

注意事项

watcher的使用

1.watcher分类

watcher主要有以下添加方式，不同添加方式监控不同的动作。（ 设置watcher时，如果对应服务端已经不存在node，watcher是不会留在服务端）

###获取节点数据，第一个方法中watch=true代表使用zookeeper创建时传入的默认watcher，监控动作sets data on the node, or deletes the node.
publicbyte[] getData(String path, booleanwatch, Stat stat)
publicbyte[] getData(finalString path, Watcher watcher, Stat stat)

###判断节点是否存在，第一个方法中watch=true代表使用zookeeper创建时传入的默认watcher，监控动作creates/delete the node or sets  the data on the node.
publicStat exists(finalString path, Watcher watcher)
publicStat exists(String path, booleanwatch)
###获取节点child，第一个方法中watch=true代表使用zookeeper创建时传入的默认watcher，监控动作deletes the node of the given path or creates/delete a child under the node
publicList<String> getChildren(finalString path, Watcher watcher)
publicList<String> getChildren(String path, booleanwatch)

2.watcher的使用注意

下面几个点需要在使用的时候多加注意

watcher是一次性的，每次处理事件之后需要重新加上。在处理和重新加上的间隔之间事件变化无法得知。

###获取的时候把watcher重新加一次
List<String> nodes = ZkUtils.getChildren(ZkUtils.ZK_PUSH_SERVER_LIST_ROOT_NODE,ServerListWatcher.getInstance());

客户端与zookeeper Disconnect时发生的事件可能会丢失。

###如果是断开重连，需要重新获取下list，以免丢失事件
if(event.getState() == KeeperState.SyncConnected) {
getAndSaveNodeList();
}

如果session过期，所有watcher就木有了。

if(event.getState() == KeeperState.Expired) {
while(true) {
try{
// 重新初始化zk连接
ZkUtils.initZk();
// 重新初始化相关zk节点
PushZkUtils.initPushZkNodes();
// 获取nodeList
getAndSaveNodeList();
// 如果成功，则break;
break;
}catch(RuntimeException e) {
AsyncLogger.getInstance().info(LogManager.SERVICE_LOG,"ServerListWatcher:Init Error!");
// 5s后重新执行
Thread.sleep(5*1000);
}
}

CONNECTIONLOSS和SESSIONEXPIRED

因为网络环境等原因，这两个异常还是会时不时出现一下，需要对这些情况有所考虑。

ConnectionLoss表示认为该连接不可用，zk的java client会自动拿着当前的sessionId去其他的zkServer建立连接。

如果在SessionTimeout的时间之内重新建立连接成功，则一切恢复正常，临时节点和watcher不会发生丢失。

如果在SessionTimeout时间之后才连接上zkServer，则zkServer会返回Session Expired，这个时候前一个session不可用，临时节点和watcher丢失，需要手动去冲新进行初始化。

ps：Zk实例化的时候会传入一个SessionTimeOut的参数，zk服务端在接收到这个参数后会和本地配置minSessionTimeout（默认2*tickTime）和maxSessionTimeout（默认20*tickTime）进行比较并进行截取。