ZooKeeper 回调机制_zk回调失效

ZooKeeper Watch机制

   一个ZooKeeper的节点可以被监控，包括这个目录中存储的数据的修改，子节点目录的变化，一旦变化可以通知设置监控的客户端。通过这个特性可以实现的功能包括配置的集中管理，集群管理，分布式锁等等。ZooKeeper Watch机制具有如下特性：

• 一次性的触发器（one-time trigger）
    无论是服务端还是客户端，一旦一个Watcher被触发，ZooKeeper都会将其从相应的存储中移除。这样的设计有效地减轻了服务端的压力。如果注册一个Watcher之后一直有效，那么针对更新非常频繁的节点，服务端会不断地向客户端发送事件通知，这无论对于网络还是服务端性能的影响都非常大
• 发送给客户端（Sent to the client）
    Watch的通知事件是从服务器发送给客户端的，是异步的，即不同的客户端收到的Watch的时间可能不同。Zookeeper 本身提供了保序性(ordering guarantee)：即客户端只有首先看到了监视事件后，才会感知到它所设置监视的 znode 发生了变化。网络延迟或者其他因素可能导致不同的客户端在不同的时刻感知某一监视事件，但是不同的客户端所看到的一切具有一致的顺序。
• 设置Watch的数据内容
    Znode改变有很多种方式，例如：节点创建，节点删除，节点改变，子节点改变等等。Zookeeper维护了两个Watch列表，一个节点数据Watch列表，另一个是子节点Watch列表。getData()和exists()设置数据Watch，getChildren()设置子节点Watch。两者选其一，可以让我们根据不同的返回结果选择不同的Watch方式，getData()和exists()返回节点的内容，getChildren()返回子节点列表。因此，setData()触发内容Watch，create()触发当前节点的内容Watch或者是其父节点的子节点Watch。delete()同时触发父节点的子节点Watch和内容Watch，以及子节点的内容Watch。

  接下来将从ZooKeeper客户端进行Watcher注册、服务端处理Watcher以及客户端回调Wathcer三方面分阶段讲解了ZooKeeper的Watcher工作机制。

服务端处理Watcher

  客户端注册Watcher的过程中最终客户端并不会将Watcher对象真正传递到服务端。那么服务端究竟是如何完成客户端的Watcher注册，又是如何来处理这个Watcher的呢？

ServerCnxn存储

服务端处理Watcher的时序图：

服务端收到来自客户端的请求后，在FinalRequestProcessor.processRequest()中会判断当前请求是否需要注册Watcher：

case OpCode.getData:{
   
     ...
     byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat, getDataRequest.getWatch()?cnxn:null);
      rsp = new GetDataResponse(b,stat);
      break;
}
1
2
3
4
5
6
7

getData请求的处理逻辑中，当getDataRequest.getWatch()为true的时候，ZooKeeper认为当前客户端请求需要进行Watcher注册，于是就会将当前的ServerCnxn对象和数据节点路径传入getData方法中去。对于标记了Watcher注册的请求，ZooKeeper会将其对应的ServerCnxn存储到WatchManager中。
  ServerCnxn是一个ZooKeeper客户端和服务器之间的连接接口，代表了一个客户端和服务器的连接，实现了Watcher的process接口，因此可以将ServerCnxn看作是一个Watcher对象。数据节点的节点路径和ServerCnxn最终会被存储在WatcherManager的watchTable和watch2Paths中。

• WatchManager是ZooKeeper服务端Watcher的管理者，其内部管理的watchTable和watch2Pashs两个存储结构，分别从两个维度对Watcher进行存储
• watchTable是从数据节点路径的粒度来托管Watcher
• watch2Paths是从Watcher的粒度来控制事件触发需要触发的数据节点

服务端触发Watcher

  NodeDataChanged事件的触发条件是“Watcher监听的对应数据节点的数据内容发生变更”，其具体实现如下：

public Stat setData（String path， byte data[] ,int version,long zxid,long time）throws KeeperException.NoNodeException{
   
       Stat s = new Stat();
       DataNode n = nodes.get(path);
       if(n == null){
   
               throw new KeeperException.NoNodeException();
       }
        byte lastdata[] = null;
   
1
2
3
4
5
6
7
8
9