zookeeper学习心得三：Watcher_综合

　　接下来我们要继续学习zookeeper的watch机制，很想了解watch中有什么用的。

1、watcher介绍

　　Zookeeper 并不是用来专门存储数据的，它的作用主要是用来维护和监控你存储的数据的状态变化。通过监控这些数据状态的变化，从而可以达到基于数据的集群管。

2、watcher实现

　　可以直接写一个继承Watcher的类，然后实现process接口方法，或者直接在实现类中定义一个Watcher对象，实现process方法
1、第一种继承Watcher

public class DataMonitor implements Watcher{public void process(WatchedEvent event) {}}
}

2、第二种自定义对象

    Watcher wh = new Watcher() {public void process(org.apache.zookeeper.WatchedEvent event) {System.out.println(event.toString());}};

3、谁可以注册谁可以触发

　　我们已经知道了watcher简单实现框架，那问题来了，谁可以注册watcher，谁可以触发watcher？

注册watcher
可以注册watcher的方法只有getData、exists、getChildren。
触发watcher
create、delete、setData。连接断开的情况下触发的watcher会丢失。
一个Watcher实例是一个回调函数，被回调一次后就被移除了。如果还需要关注数据的变化，需要再次注册watcher。
New ZooKeeper时注册的watcher叫default watcher，它不是一次性的，只对client的连接状态变化作出反应。

3、注册和watcher的对应表
　　如图所示：
（1）什么样的操作引发什么样的事件

（2）事件类型与watcher的对应关系

（3）操作与watcher的对应关系

通过调用方法注册watcher，然后通过方法来触发这个watcher，就可以得到通知消息。

4、WatchedEvent event

　　方法public void process(WatchedEvent event) 中有一个event对象，这个对象里面存放着触发后的相关属性，很多都跟触发的发给方法有关，接下来我们来了解一下。
（1）EventType.NodeChildrenChanged;
触发条件：所关注的节点的子节点有变化。这里说的变化是指子节点的个数和组成，具体到子节点内容的变化是不会通知的。
（2）EventType.NodeCreated;
所关注的节点被创建。
（3）EventType.NodeDataChanged;
触发条件：所关注的节点的内容有更新。注意，这个地方说的内容是指数据的版本号dataVersion。因此，即使使用相同的数据内容来更新，还是会收到这个事件通知的。无论如何，调用了更新接口，就一定会更新dataVersion的。
（4）EventType.NodeDeleted;
所关注的节点被删除。
（5）EventType.None;
触发条件：一般是在与服务器断开连接的时候，客户端会收到这个事件
（6）KeeperState.SyncConnected
（7）KeeperState.Disconnected(0)
此时客户端处于断开连接状态，和ZK集群都没有建立连接

　　通过这些返回的类型，我们可以执行相应的操作，但是要记得watcher是一次性的，要想继续监控节点，就必须继续注册watcher。

代码模拟


import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;public class DistributedClient {  // 超时时间 private static final int SESSION_TIMEOUT = 50000;  // zookeeper server列表 private String hosts = "218.193.126.190:2181";  private String groupNode = "locks";  private String subNode = "sub";  private ZooKeeper zk;  // 当前client创建的子节点 private String thisPath;  // 当前client等待的子节点 private String waitPath;  private CountDownLatch latch = new CountDownLatch(1);  public Watcher getWatcher(final String msg) {return new Watcher() {@Overridepublic void process(WatchedEvent event) {try {// 连接建立时, 打开latch, 唤醒wait在该latch上的线程if (event.getState() == KeeperState.SyncConnected) {latch.countDown();}// 发生了waitPath的删除事件if (event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {  // 确认thisPath是否真的是列表中的最小节点 List<String> childrenNodes = zk.getChildren("/" + groupNode, false);  String thisNode = thisPath.substring(("/" + groupNode + "/").length());  // 排序 Collections.sort(childrenNodes);  int index = childrenNodes.indexOf(thisNode);  if (index == 0) {  // 确实是最小节点 doSomething();  } else {      // 否则说明waitPath是由于出现异常而挂掉的 // 更新waitPath waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);  // 重新注册监听, 并判断此时waitPath是否已删除 if (zk.exists(waitPath, true) == null) {  doSomething();  }  }  }  } catch (Exception e) {e.printStackTrace();}}};}/** * 连接zookeeper */  public void connectZookeeper() throws Exception {  System.out.println("进入连接zookeeper函数中");zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher(){public void process(WatchedEvent event) {try {// 连接建立时, 打开latch, 唤醒wait在该latch上的线程if (event.getState() == KeeperState.SyncConnected) {latch.countDown();}// 发生了waitPath的删除事件if (event.getType() == EventType.NodeDeleted&& event.getPath().equals(waitPath)) {System.out.println("进入waitpath路径的删除时间中了");doSomething();System.out.println("执行节点被删除的通知时间：dosomething函数");}} catch (Exception e) {e.printStackTrace();}}});// 等待连接建立 latch.await();  // 创建子节点 thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,  CreateMode.EPHEMERAL_SEQUENTIAL);  System.out.println("thispath---"+thisPath);// wait一小会, 让结果更清晰一些 Thread.sleep(10);  // 注意, 没有必要监听"/locks"的子节点的变化情况 List<String> childrenNodes = zk.getChildren("/" + groupNode, false);  System.out.println("开始判断节点个数");// 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁 if (childrenNodes.size() == 1) {  System.out.println("孩子节点个数等于1");doSomething();  } else {  System.out.println("孩子节点个数大于1");String thisNode = thisPath.substring(("/" + groupNode + "/").length());  System.out.println("当前节点--"+thisNode);// 排序 ------------后面才是重点的Collections.sort(childrenNodes);  int index = childrenNodes.indexOf(thisNode);  if (index == -1) {  // never happened } else if (index == 0) {  // inddx == 0, 说明thisNode在列表中最小, 当前client获得锁 doSomething();  } else {  // 获得排名比thisPath前1位的节点 ，如果当前的节点不是序号最低的哪一个，就去当前序号的前一个，this.waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);  // 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法 zk.getData(waitPath, true, new Stat());  }  }  }  private void doSomething() throws Exception {  try {  System.out.println("gain lock: " + thisPath);  Thread.sleep(2000);  // do something } finally {  System.out.println("finished: " + thisPath);  // 将thisPath删除, 监听thisPath的client将获得通知 // 相当于释放锁 zk.delete(this.thisPath, -1);  }  }  public static void main(String[] args) throws Exception {  for (int i = 0; i < 10; i++) {  new Thread() {  public void run() {  try {  DistributedClient dl = new DistributedClient();  dl.connectZookeeper();  } catch (Exception e) {  e.printStackTrace();  }  }  }.start();  }  Thread.sleep(Long.MAX_VALUE);  }  
}

仅供参考。