ZooKeeper Watcher机制

小咪咪 2023-05-29 03:55 7阅读 0赞

要获取一个节点的数据变化，通常有两种方案：

*  轮询：实时性差，资源利用率低，浪费大量的 CPU 和 带宽；
 *  事件：数据状态变更时，主动通知 Client，节省大量资源；ZK Server 通过 watcher 监视点，来实现事件通知。

而zk使用的是后一种方案，一个zk服务器集群上的节点Znode发生变化（Znode本身的增加，删除，修改，以及子Znode的变化）就可以通过Watch机制通知到客户端。那么要实现Watch，就必须实现org.apache.zookeeper.Watcher接口，并且将实现类的对象传入到可以Watch的方法中。Zookeeper中所有读操作（getData()，getChildren()，exists()）都可以设置Watch选项。Watch事件具有one-time trigger（一次性触发）的特性，如果Watch监视的Znode有变化，那么就会通知设置该Watch的客户端。这个功能是zookeeper对于应用最重要的特性，通过这个特性可以实现的功能包括配置的集中管理，集群管理，分布式锁等等。Watch的整体流程如下图所示，客户端先向ZooKeeper服务端成功注册想要监听的节点状态，同时客户端本地会存储该监听器相关的信息在WatchManager中，当ZooKeeper服务端监听的数据状态发生变化时，ZooKeeper就会主动通知发送相应事件信息给相关会话客户端，客户端就会在本地响应式的回调相关Watcher的Handler。

### zookeeper机制的特点 ###

1.  一次性触发器：客户端在Znode设置了Watch时，如果Znode内容发生改变，那么客户端就会获得Watch事件。例如：客户端设置getData("/znode1", true)后，如果/znode1发生改变或者删除，那么客户端就会得到一个/znode1的Watch事件，但是/znode1再次发生变化，那客户端是无法收到Watch事件的，除非客户端设置了新的Watch。
2.  发送至客户端，串行执行：Watch事件是异步发送到Client。Zookeeper可以保证客户端发送过去的更新顺序是有序的。例如：某个Znode没有设置watcher，那么客户端对这个Znode设置Watcher发送到集群之前，该客户端是感知不到该Znode任何的改变情况的。换个角度来解释：由于Watch有一次性触发的特点，所以在服务器端没有Watcher的情况下，Znode的任何变更就不会通知到客户端。不过，即使某个Znode设置了Watcher，且在Znode有变化的情况下通知到了客户端，但是在客户端接收到这个变化事件，但是还没有再次设置Watcher之前，如果其他客户端对该Znode做了修改，这种情况下，Znode第二次的变化客户端是无法收到通知的。这可能是由于网络延迟或者是其他因素导致，所以我们使用Zookeeper不能期望能够监控到节点每次的变化。Zookeeper只能保证最终的一致性，而无法保证强一致性。
3.  设置watch的数据内容：Znode改变有很多种方式，例如：节点创建，节点删除，节点改变，子节点改变等等。Zookeeper维护了两个Watch列表，一个节点数据Watch列表，另一个是子节点Watch列表。getData()和exists()设置数据Watch，getChildren()设置子节点Watch。两者选其一，可以让我们根据不同的返回结果选择不同的Watch方式，getData()和exists()返回节点的内容，getChildren()返回子节点列表。因此，setData()触发内容Watch，create()触发当前节点的内容Watch或者是其父节点的子节点Watch。delete()同时触发父节点的子节点Watch和内容Watch，以及子节点的内容Watch。
4.  轻量级设计：WatchedEvent 是 ZooKeeper 整个 Watcher 通知机制的最小通知单元，这个数据结构中只包含三部分的内容：通知状态、事件类型和节点路径。也就是说，Watcher 通知非常简单，只会告诉客户端发生了事件，而不会说明事件的具体内容

### 监视点的生命周期： ###

*  跟 Session 生命周期绑定：Session 关闭或超时，监视点会自动清除
 *  监视点只保存在内存中，不会持久化到磁盘
 *  Client 重连有效：Client 侧保存一份监视点信息，Client 重连到另一个 Server 后，会向 Server 发送监视点列表，重新设置监视点

### 具体实现 ###

#### 1. 回调 ####

Watcher机制实现的基础是回调，所以我们看一下基础的回调：

class Caller{
      private Call c;
    
      public void setCall(Call c){
        this.c = c;
      }
      public void call(){
        this.c.method();
      }
    }
    
    interface Call{
      public void methid();
    }
    
    
    class CallImpl implements Call{
      public void mehtod(){
        System.out.println("call");
      }
    }
    
    public class Test{
      public static void main(){
        Caller caller = new Caller();
        caller.setCall(new CallImpl());
        caller.call();
      }
    }

#### 2. zk处理流程： ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1_size_16_color_FFFFFF_t_70]

由图看出，zk的watcher由客户端，客户端WatchManager，zk服务器组成。整个过程涉及了消息通信及数据存储。

*  zk客户端向zk服务器注册watcher的同时，会将watcher对象存储在客户端的watchManager。
 *  Zk服务器触发watcher事件后，会向客户端发送通知，客户端线程从watchManager中回调watcher执行相应的功能。

Watcher 的理念是启动一个客户端去接收从 ZooKeeper 服务端发过来的消息并且同步地处理这些信息。ZooKeeper 的 Java API 提供了公共接口 Watcher，具体操作类通过实现这个接口相关的方法来实现从所连接的 ZooKeeper 服务端接收数据。如果要处理这个消息，需要为客户端注册一个 CallBack（回调）对象。Watcher 接口定义在 org.apache.zookeeper 包里面，代码如下：

public interface Watcher {
      abstract public void process(WatchedEvent event);
    }

其watcher是监听数据发送了某些变化，那就一定会有对应的事件类型，和状态类型，如下所示：

1.  事件类型：（znode节点相关的）

*  EventType.NodeCreated
 *  EventType.NodeDataChanged
 *  EventType.NodeChildrenChanged
 *  EventType.NodeDeleted

1.  状态类型：（是跟客户端实例相关的）

*  KeeperState.Disconnected
 *  KeeperState.SyncConnected
 *  KeeperState.AuthFailed
 *  KeeperState.Expired

客户端的请求基本都是在 ClientCnxn 里面进行操作，当收到请求后，客户端会对当前客户端请求进行标记，将其设置为使用 Watcher 监听，同时会封装一个 Watcher 的注册信息 WatchRegistration 对象，用于暂时保存数据节点的路径和 Watcher 的对应关系。

在服务端DataTree会托管两个WatchManager，分别是dataWatches和childWatches，分别对应数据变更Watcher和子节点变更Watcher。watchManager，Zk服务器端Watcher的具体管理者，从两个维度维护watcher，watchTable从数据节点的粒度来维护和watch2Paths从watcher的粒度来维护，它负责watcher事件的触发

class WatchManager {
 private final Map<String, Set<Watcher>> watchTable =new HashMap<String, Set<Watcher>>();
 
 private final Map<Watcher, Set<String>> watch2Paths = new HashMap<Watcher, Set<String>>();
 }

在看addWatch方法，watcher就是注册在这两个变量之上的，方便操作：

public synchronized void addWatch(String path, Watcher watcher) {
 HashSet<Watcher> list = watchTable.get(path);
 if (list == null) {
 // don't waste memory if there are few watches on a node
 // rehash when the 4th entry is added, doubling size thereafter
 // seems like a good compromise
 list = new HashSet<Watcher>(4);
 watchTable.put(path, list);
 }
 list.add(watcher);
 
 HashSet<String> paths = watch2Paths.get(watcher);
 if (paths == null) {
 // cnxns typically have many watches, so use default cap here
 paths = new HashSet<String>();
 watch2Paths.put(watcher, paths);
 }
 paths.add(path);
 }

这个getChildren的添加事件：

public ArrayList<String> getChildren(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException {
 DataNodeV1 n = nodes.get(path);
 if (n == null) {
 throw new KeeperException.NoNodeException();
 }
 synchronized (n) {
 ArrayList<String> children = new ArrayList<String>();
 children.addAll(n.children);
 if (watcher != null) {
 childWatches.addWatch(path, watcher);
 }
 return children;
 }
 }

到这里相当于上图中的步骤1和步骤2完成了；那么步骤3，通知是怎么做的呢，举个例子：session的变化，Zookeeper客户端和服务端维持一个长连接，每隔10s向服务端发送一个心跳，服务端返回客户端一个响应。这就是一个Session连接，拥有全局唯一的session id。Session连接通常是一直有效，如果因为网络原因断开了连接，客户端会使用相同的session id进行重连。由于服务端保留了session的各种状态，尤其是各种瞬时节点是否删除依赖于session是否失效。通常客户端主动关闭连接认为是一次session失效。另外也有可能因为其它未知原因，例如网络超时导致的session失效问题。在服务端看来，无法区分session失效是何种情况，一次一旦发生session失效，一定时间后就会将session持有的所有watcher以及瞬时节点删除。在这种情况下触发了。其他还有Create、Delete、NodeChange等，在这种情况下会去触发 WatchManager 的 triggerWatch 方法：

public Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {
 WatchedEvent e = new WatchedEvent(type, KeeperState.SyncConnected, path);
 //将事件类型（EventType）、通知状态（WatchedEvent）、节点路径封装成一个 WatchedEvent 对象
 HashSet<Watcher> watchers;
 synchronized (this) {
 /**
 * 根据数据节点的节点路径从 watchTable 里面取出对应的Watcher。如果没有找到 Watcher 对象，
 * 说明没有任何客户端在该数据节点上注册过 Watcher，直接退出。如果找打了 Watcher 就将其提取出来，
 * 同时会直接从watchTable和watch2Paths里删除 Watcher，即 Watcher 是一次性的，触发一次就失效了。
 */
 watchers = watchTable.remove(path);
 for (Watcher w : watchers) {
 HashSet<String> paths = watch2Paths.get(w);
 }
 }
 for (Watcher w : watchers) {
 if (supress != null && supress.contains(w)) {
 continue;
 }
 /**
 * 对于需要注册 Watcher 的请求，ZooKeeper 会把请求对应的恶 ServerCnxn 作为一个 Watcher 存储，
 * 所以这里调用的 process 方法实质上是 ServerCnxn 的对应方法
 */
 
 w.process(e);
 }
 return watchers;
 }

这样就完成了步骤4的调用，总结一下就是：注册一个watcher并存储，然后在触发的时候取出watcher继续调用。在看下总体流程：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1_size_16_color_FFFFFF_t_70 1]

然后看一个应用例子：

package bhz.zookeeper.watcher;
 
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import javafx.scene.shape.Path;
 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;
 
 /**
 * Zookeeper Wathcher 
 * This class is a Watcher class (implemented by the org.apache.zookeeper.Watcher class)
 * @author（alienware）
 * @since 2019-2-19
 */
 public class ZooKeeperWatcher implements Watcher {
 
 /** Define atomic variables */
 AtomicInteger seq = new AtomicInteger();
 /** Define session expiration time */
 private static final int SESSION_TIMEOUT = 10000;
 /** zookeeper server address */
 private static final String CONNECTION_ADDR = "192.168.1.24:2181";
 /** zk parent path setting */
 private static final String PARENT_PATH = "/p";
 /** zk subpath setting */
 private static final String CHILDREN_PATH = "/p/c";
 /** Enter the logo */
 private static final String LOG_PREFIX_OF_MAIN = "【Main】";
 /** zk variable */
 private ZooKeeper zk = null;
 /** Used to wait for the zookeeper connection to be established. Notify the blocking program to continue to execute downwards */
 private CountDownLatch connectedSemaphore = new CountDownLatch(1);
 
 /**
 * Create a ZK connection
 * @param connectAddr ZK server address list
 * @param sessionTimeout Session timeout
 */
 public void createConnection(String connectAddr, int sessionTimeout) {
 this.releaseConnection();
 try {
 //this means passing the current object to it (that is, the new ZooKeeperWatcher() instance object instantiated in the main function)
 zk = new ZooKeeper(connectAddr, sessionTimeout, this);
 System.out.println(LOG_PREFIX_OF_MAIN + "Start connecting to the ZK server");
 connectedSemaphore.await();
 } catch (Exception e) {
 e.printStackTrace();
 }
 }
 
 /**
 * Close ZK connection
 */
 public void releaseConnection() {
 if (this.zk != null) {
 try {
 this.zk.close();
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 }
 
 /**
 * Create a node
 * @param path node path
 * @param data data content
 * @return 
 */
 public boolean createPath(String path, String data, boolean needWatch) {
 try {
 / / Set up monitoring (since zookeeper monitoring is one-time, so you must set up monitoring each time)
 this.zk.exists(path, needWatch);
 System.out.println(LOG_PREFIX_OF_MAIN + "The node was created successfully, Path: " + 
 This.zk.create( /**path*/ 
 path, 
 /**Data*/
 data.getBytes(), 
 /**All visible*/
 Ids.OPEN_ACL_UNSAFE, 
 /**Permanent storage*/
 CreateMode.PERSISTENT ) + 
 ", content: " + data);
 } catch (Exception e) {
 e.printStackTrace();
 return false;
 }
 return true;
 }
 
 /**
 * Read the specified node data content
 * @param path node path
 * @return
 */
 public String readData(String path, boolean needWatch) {
 try {
 System.out.println("Read Data Operation...");
 return new String(this.zk.getData(path, needWatch, null));
 } catch (Exception e) {
 e.printStackTrace();
 return "";
 }
 }
 
 /**
 * Update the specified node data content
 * @param path node path
 * @param data data content
 * @return
 */
 public boolean writeData(String path, String data) {
 try {
 System.out.println(LOG_PREFIX_OF_MAIN + "Update data successfully, path:" + path + ", stat: " +
 this.zk.setData(path, data.getBytes(), -1));
 } catch (Exception e) {
 e.printStackTrace();
 return false;
 }
 return true;
 }
 
 /**
 * Delete the specified node
 * 
 * @param path
 * node path
 */
 public void deleteNode(String path) {
 try {
 this.zk.delete(path, -1);
 System.out.println(LOG_PREFIX_OF_MAIN + "Delete node successfully, path:" + path);
 } catch (Exception e) {
 e.printStackTrace();
 }
 }
 
 /**
 * Determine if the specified node exists
 * @param path node path
 */
 public Stat exists(String path, boolean needWatch) {
 try {
 return this.zk.exists(path, needWatch);
 } catch (Exception e) {
 e.printStackTrace();
 return null;
 }
 }
 
 /**
 * Get child nodes
 * @param path node path
 */
 private List<String> getChildren(String path, boolean needWatch) {
 try {
 System.out.println("Read child node operation...");
 return this.zk.getChildren(path, needWatch);
 } catch (Exception e) {
 e.printStackTrace();
 return null;
 }
 }
 
 /**
 * Delete all nodes
 */
 public void deleteAllTestPath(boolean needWatch) {
 if(this.exists(CHILDREN_PATH, needWatch) != null){
 this.deleteNode(CHILDREN_PATH);
 }
 if(this.exists(PARENT_PATH, needWatch) != null){
 this.deleteNode(PARENT_PATH);
 } 
 }
 
 /**
 * Processing after receiving a Watcher notification from Server.
 */
 @Override
 public void process(WatchedEvent event) {
 
 System.out.println("Enter process . . . event = " + event);
 
 try {
 Thread.sleep(200);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 
 if (event == null) {
 return;
 }
 
 // Connection Status 
 KeeperState keeperState = event.getState();
 // event type
 EventType eventType = event.getType();
 // affected path
 String path = event.getPath();
 // The number of times the atomic object seq records into the process
 String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";
 
 System.out.println(logPrefix + "received Watcher notification");
 System.out.println(logPrefix + "Connection Status: \t" + keeperState.toString());
 System.out.println(logPrefix + "Event Type:\t" + eventType.toString());
 
 if (KeeperState.SyncConnected == keeperState) {
 // The first successful connection to the ZK server
 if (EventType.None == eventType) {
 System.out.println(logPrefix + "Successfully connected to the ZK server");
 connectedSemaphore.countDown();
 } 
 / / Create a node
 else if (EventType.NodeCreated == eventType) {
 System.out.println(logPrefix + "node creation");
 try {
 Thread.sleep(100);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 } 
 / / Update node
 else if (EventType.NodeDataChanged == eventType) {
 System.out.println(logPrefix + "node data update");
 try {
 Thread.sleep(100);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 } 
 / / Update the child node
 else if (EventType.NodeChildrenChanged == eventType) {
 System.out.println(logPrefix + "child node change");
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 } 
 / / Delete the node
 else if (EventType.NodeDeleted == eventType) {
 System.out.println(logPrefix + "node" + path + "deleted");
 }
 else ;
 } 
 else if (KeeperState.Disconnected == keeperState) {
 System.out.println(logPrefix + "disconnected from the ZK server");
 } 
 else if (KeeperState.AuthFailed == keeperState) {
 System.out.println(logPrefix + "Permission check failed");
 } 
 else if (KeeperState.Expired == keeperState) {
 System.out.println(logPrefix + "session invalid");
 }
 else ;
 
 System.out.println("--------------------------------------------");
 
 }
 
 /**
 * Method Name: Test Zookeeper Monitoring 
 * Overview: Main test watch function 
 * @param args
 * @throws Exception
 */
 public static void main(String[] args) throws Exception {
 
 //Create watcher //The current client can be called a watcher observer role
 ZooKeeperWatcher zkWatch = new ZooKeeperWatcher();
 / / Create a connection 
 zkWatch.createConnection(CONNECTION_ADDR, SESSION_TIMEOUT);
 //System.out.println(zkWatch.zk.toString());
 
 Thread.sleep(1000);
 
 // clean up the node
 // zkWatch.deleteAllTestPath(false);
 
 //-----------------Step 1: Create a parent node /p ---------------------- --//
 // Create a parent node
 /**
 * data: the content of the node is the current time millisecond value, needWATCH: whether it needs to be monitored
 */
 if (zkWatch.createPath(PARENT_PATH, System.currentTimeMillis() + "", true)) {
 
 Thread.sleep(1000);
 
 //----------------- Step 2: Read the difference between the node /p and the child node under the /p node (getChildren) ------- -------//
 // read the data
 // zkWatch.readData(PARENT_PATH, true);
 // You can also do this, indicating that the current node needs to continue watching, and the update data operation will be monitored.
 zkWatch.exists(PARENT_PATH,true);
 
 // Read child nodes (monitor childNodeChange event)
 // The listener here still listens for PARENT_PATH, but there is a trigger for the child node here, the trigger here is still the parent node.
 // As long as the child node changes [delete, modify, increase] then it will trigger, only trigger: NodeChildrenChanged
 zkWatch.getChildren(PARENT_PATH, true);
 
 / / Update the data, because the above watch, so the update operation will be monitored, when the update is over, the above watch relationship will disappear
 zkWatch.writeData(PARENT_PATH, System.currentTimeMillis() + "");
 
 Thread.sleep(1000);
 // Create a child node
 // If needWatch is false, it simply creates a child node and does not trigger any watch:NodeCreated
 zkWatch.createPath(CHILDREN_PATH, System.currentTimeMillis() + "", true);
 
 
 //-----------------Step 3: Create a trigger for the child node --------------//
 // zkWatch.getChildren(CHILDREN_PATH, true);
 // zkWatch.createPath(CHILDREN_PATH + "/c1", System.currentTimeMillis() + "", true);
 // zkWatch.getChildren(CHILDREN_PATH + "/c1", true);
 // zkWatch.createPath(CHILDREN_PATH + "/c1/c2", System.currentTimeMillis() + "", true);
 
 //-----------------Step 4: Update the trigger of the child node data --------------//
 // Before we make the changes, we need to watch this node:
 Thread.sleep(1000);
 zkWatch.readData(CHILDREN_PATH, true);
 zkWatch.writeData(CHILDREN_PATH, System.currentTimeMillis() + "");
 
 }
 
 Thread.sleep(10000);
 // clean up the node
 zkWatch.deleteAllTestPath(true);
 
 Thread.sleep(10000);
 zkWatch.releaseConnection();
 
 }
 }

参考地址：

https://juejin.im/post/5b0784ab6fb9a07abf72f098

https://ningg.top/zookeeper-lesson-6-zookeeper-watcher/

https://qinzhaokun.github.io/2017/07/12/zookeeperWatch%E6%9C%BA%E5%88%B6/

https://www.ibm.com/developerworks/cn/opensource/os-cn-apache-zookeeper-watcher/index.html

https://blog.csdn.net/qianshangding0708/article/details/50084155

https://mastertf.github.io/2017/04/24/zookeeper%E4%B9%8B%E7%9B%91%E5%90%AC%E4%BA%8B%E4%BB%B6/

https://www.jianshu.com/p/4c071e963f18

http://walkerdu.com/2019/02/11/zookeeper\_basic/

http://www.programmersought.com/article/4519691445/

https://books.google.com.kh/books?id=OWQdAgAAQBAJ&pg=PT95&lpg=PT95&dq=zk+watcher+mechanism&source=bl&ots=14FOpxp56Q&sig=ACfU3U0Yfx\_0bzJYA4Djol\_v9KAGDdN\_Wg&hl=zh-CN&sa=X&redir\_esc=y\#v=onepage&q&f=false

https://www.owenzhang.net/blog/121.html

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191104093741743.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191104093805478.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1,size_16,color_FFFFFF,t_70