zookeeper watches

Watches

所有的读操作（getData(),getChildren(),exists()）都可以设置一个watch作为附加的操作

Watch的关键特性有

• 只触发一次（one-timetrigger）

Watch当数据变化时，只会被触发一次，之后就会被移除，当数据再次变化时，不会再触发此watch

• 发送到客户端（sentto the client）

通知客户端是异步的，客户端只会在它收到watch event时才会知道znode改变.通知客户端的过程可能会发生异常，或者网络延迟，但zookeeper会保证通知的一致性

• Watch为那些数据znode设置（the data for which the watch was set）

getData(),exists()设置数据监控，getChildren()设置孩子监控，

setData()会触发数据监控，成功create（）会触发创建的znode数据监控和父zonde的孩子监控，成功的delete()zonde的数据监控和孩子监控、还有父zonde的孩子监控

Zookeeper 对watches一些保证

• Zookeeper客户端保证所有的事情都是有序的（events,watches,异步回复）

• 客户端先看到监听的事件，先于所监听的znode新的数据

• 监听事件的顺序对应zookeeper服务发生的更新时间顺序

  关于监听器必须注意的

• 监听器只触发一次

• 因为监听是一次性的，并且在获取事件通知和发送新的请求设置监听器之间有延迟，所以不能完全相信每个事件的变化就是zonde当前的状态，有可能发生了多次变化之后你才收到监听事件的回复

• 同一个监听对象只会被一个通知事件触发，比如同一个监听对象被注册到一个exists和

GetDate，当删除这个节点的时候，监听对象只会被删除事件触发一次

• 当和服务器断开后，客户端不会获得任何监听器，直到重建连接，

以下是我根据官网例子简化的实例，大家可以动手实践以下，理解一下zookeeper的监听器的操作

package personal.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;

/**
* comments()
*
* @author:czh ,mail:[email protected]
* @date 2015/11/4
*/
@Slf4j
public class Executor implementsWatcher,Runnable,AsyncCallback.StatCallback {
Integer mutex=0;
String znode;

ZooKeeper zk;

String filename;

String exec[];


public Executor(StringhostPort, String znode) throws KeeperException, IOException,InterruptedException {
this.znode=znode;
zk = new ZooKeeper(hostPort, 3000, this);
zk.exists(znode,true, this, null);
}

/**
* @param args
*/
public static void main(String[] args) {
String hostPort ="192.168.83.3:2181";
// String hostPort="172.26.7.23:2181";
String znode = "/dubbo";
try {
new Executor(hostPort,znode).run();
} catch (Exceptione) {
e.printStackTrace();
}
}

/***************************************************************************
* We do process any eventsourselves, we just need to forward them on.
*
// * @seeorg.apache.zookeeper.Watcher#process(org.apache.zookeeper.proto.WatcherEvent)
*/
public void process(WatchedEventevent) {
log.info("event.getPath():" + event.getPath() + "event.getState():" + event.getState() + "event.getType():" + event.getType());
}

public void run(){
try {
synchronized (mutex) {
while (true) {
mutex.wait();
}
}
} catch (InterruptedExceptione) {
e.printStackTrace();
}
}

/**
* @param rc
* @param path
* @param ctx
* @param stat
*/
@Override
public void proce***esult(int rc, String path, Object ctx, Stat stat) {
log.info("proce***esult===========begin");
boolean exists;
log.info(String.valueOf(rc));
switch (rc) {
case KeeperException.Code.Ok:
log.info("监听器设置成功");
break;
case KeeperException.Code.NoNode:
log.info("znode["+znode+"]不存在");
break;
case KeeperException.Code.SessionExpired:
case KeeperException.Code.NoAuth:
return;
default:
// Retry errors
zk.exists(znode, true, this, null);
return;
}
log.info("proce***esult===========end");
}
}