zk的watcher机制的实现

﹏ヽ暗。殇╰゛Y 2022-04-24 09:56 181阅读 0赞

# [zk的watcher机制的实现][zk_watcher] #

**目录**

*  [ZooKeeper Watcher 机制][ZooKeeper Watcher]
    
     *  [集群状态监控示例][Link 1]
     *  [回调函数][Link 2]
     *  [原理及源代码解释][Link 3]
 *  [ZooKeeper Watcher 特性总结][ZooKeeper Watcher 1]
 *  [结束语][Link 4]

--------------------

转载：https://www.ibm.com/developerworks/cn/opensource/os-cn-apache-zookeeper-watcher/

http://blog.csdn.net/tycoon1988/article/details/38405101

可以设置观察的操作：exists,getChildren,getData

可以触发观察的操作：create,delete,setData

zookeeper观察机制;

服务端只存储事件的信息,客户端存储事件的信息和Watcher的执行逻辑.ZooKeeper客户端是线程安全的每一个应用只需要实例化一个ZooKeeper客户端即可，同一个ZooKeeper客户端实例可以在不同的线程中使用。ZooKeeper客户端会将这个Watcher对应Path路径存储在ZKWatchManager中,同时通知ZooKeeper服务器记录该Client对应的Session中的Path下注册的事件类型。当ZooKeeper服务器发生了指定的事件后,ZooKeeper服务器将通知ZooKeeper客户端哪个节点下发生事件类型，ZooKeeper客户端再从ZKWatchManager中找到相应Path，取出相应watcher引用执行其回调函数process。

源码分析如：

zookeeper.java

wcb = new DataWatchRegistration(watcher, clientPath);//将观察者引用和节点路径绑定

ZKWatchManager.java

public void register(int rc) \{

if (shouldAddWatch(rc)) \{

Map<String, Set<Watcher>> watches = getWatches(rc);

synchronized(watches) \{

Set<Watcher> watchers = watches.get(clientPath);

if (watchers == null) \{

watchers = new HashSet<Watcher>();

watches.put(clientPath, watchers);//key为path,value为watcher引用

watchers.add(watcher);

znode存储数据量最大：jute.maxbuffer默认1M

通知注意点：

1.一次性触发器（可以通过定时调用方法 get() 或者 exists()获取触发事件，手动访问触发，）  
client在一个节点上设置watch，随后节点内容改变，client将获取事件。当节点内容再次改变，client不会获取这个事件，除非它又执行了一次读操作并设置watch

2.发送至client，watch事件延迟  
watch事件异步发送至观察者。比如说client执行一次写操作，节点数据内容发生变化，操作返回后，而watch事件可能还在发往client的路上。这种情况下，zookeeper提供有序保证：client不会得知数据变化，直到它获取watch事件。网络延迟或其他因素可能导致不同client在不同时刻获取watch事件和操作返回值。

分布式系统从根本上来说就是不同节点上的进程并发执行，并且相互之间对进程的行为进行协调处理的过程。不同节点上的进程互相协调行为的过程叫做分布式同步。许多分布式系统需要一个进程作为任务的协调者，执行一些其他进程并不执行的特殊的操作，一般情况下哪个进程担当任务的协调者都无所谓，但是必须有一个进程作为协调者，自动选举出一个协调者的过程就是分布式选举。ZooKeeper 正是为了解决这一系列问题而生的。这一讲我们来谈谈 Watcher 机制，首先介绍一个监控示例，然后我们再来聊聊 Watcher 机制原理。

[回到顶部][Link 5]

## ZooKeeper Watcher 机制 ##

### 集群状态监控示例 ###

为了确保集群能够正常运行，ZooKeeper 可以被用来监视集群状态，这样就可以提供集群高可用性。使用 ZooKeeper 的瞬时（ephemeral）节点概念可以设计一个集群机器状态检测机制：

1. 每一个运行了 ZooKeeper 客户端的生产环境机器都是一个终端进程，我们可以在它们连接到 ZooKeeper 服务端后在 ZooKeeper 服务端创建一系列对应的瞬时节点，可以用/hostname 来进行区分。

2. 这里还是采用监听（Watcher）方式来完成对节点状态的监视，通过对/hostname 节点的 NodeChildrenChanged 事件的监听来完成这一目标。监听进程是作为一个独立的服务或者进程运行的，它覆盖了 process 方法来实现应急措施。

3. 由于是一个瞬时节点，所以每次客户端断开时 znode 会立即消失，这样我们就可以监听到集群节点异常。

4.NodeChildrenChanged 事件触发后我们可以调用 getChildren 方法来知道哪台机器发生了异常。

清单 1.ClusterMonitor 类

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 </td> 
 <td> <code>import java.io.IOException;</code> <code>import java.util.List;</code> <code>import org.apache.zookeeper.CreateMode;</code> <code>import org.apache.zookeeper.KeeperException;</code> <code>import org.apache.zookeeper.WatchedEvent;</code> <code>import org.apache.zookeeper.Watcher;</code> <code>import org.apache.zookeeper.ZooDefs.Ids;</code> <code>import org.apache.zookeeper.ZooKeeper;</code> &nbsp; <code>public class ClusterMonitor implements Runnable{ </code> <code>&nbsp;</code><code>private static String membershipRoot = "/Members";</code> <code>&nbsp;</code><code>private final Watcher connectionWatcher;</code> <code>&nbsp;</code><code>private final Watcher childrenWatcher;</code> <code>&nbsp;</code><code>private ZooKeeper zk;</code> <code>&nbsp;</code><code>boolean alive = true;</code> <code>&nbsp;</code><code>public ClusterMonitor(String HostPort) throws IOException,InterruptedException,KeeperException{ </code> <code>&nbsp;</code><code>connectionWatcher = new Watcher(){ </code> <code>&nbsp;</code><code>@Override</code> <code>&nbsp;</code><code>public void process(WatchedEvent event) { </code> <code>&nbsp;</code><code>// TODO Auto-generated method stub</code> <code>&nbsp;</code><code>if(event.getType() == Watcher.Event.EventType.None &amp;&amp; </code> <code>&nbsp;</code><code>event.getState() == Watcher.Event.KeeperState.SyncConnected){ </code> <code>&nbsp;</code><code>System.out.println("\nconnectionWatcher Event Received:%s"+event.toString());</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>} </code> <code>&nbsp;</code><code>};</code> <code>&nbsp;</code>&nbsp; <code>&nbsp;</code><code>childrenWatcher = new Watcher(){ </code> <code>&nbsp;</code><code>@Override</code> <code>&nbsp;</code><code>public void process(WatchedEvent event) { </code> <code>&nbsp;</code><code>// TODO Auto-generated method stub</code> <code>&nbsp;</code><code>System.out.println("\nchildrenWatcher Event Received:%s"+event.toString());</code> <code>&nbsp;</code><code>if(event.getType()==Event.EventType.NodeChildrenChanged){ </code> <code>&nbsp;</code><code>try{ </code> <code>&nbsp;</code><code>//Get current list of child znode and reset the watch</code> <code>&nbsp;</code><code>List&lt;</code><code>String</code><code>&gt; children = zk.getChildren(membershipRoot, this);</code> <code>&nbsp;</code><code>System.out.println("Cluster Membership change,Members: "+children);</code> <code>&nbsp;</code><code>}catch(KeeperException ex){ </code> <code>&nbsp;</code><code>throw new RuntimeException(ex);</code> <code>&nbsp;</code><code>}catch(InterruptedException ex){ </code> <code>&nbsp;</code><code>Thread.currentThread().interrupt();</code> <code>&nbsp;</code><code>alive = false;</code> <code>&nbsp;</code><code>throw new RuntimeException(ex);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>};</code> <code>&nbsp;</code>&nbsp; <code>&nbsp;</code><code>zk = new ZooKeeper(HostPort,2000,connectionWatcher);</code> <code>&nbsp;</code><code>//Ensure the parent znode exists</code> <code>&nbsp;</code><code>if(zk.exists(membershipRoot, false) == null){ </code> <code>&nbsp;</code><code>zk.create(membershipRoot, "ClusterMonitorRoot".getBytes(), </code> <code>&nbsp;</code><code>Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>//Set a watch on the parent znode</code> <code>&nbsp;</code><code>List&lt;</code><code>String</code><code>&gt; children = zk.getChildren(membershipRoot, childrenWatcher);</code> <code>&nbsp;</code><code>System.err.println("Members:"+children);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code>&nbsp; <code>&nbsp;</code><code>public synchronized void close(){ </code> <code>&nbsp;</code><code>try{ </code> <code>&nbsp;</code><code>zk.close();</code> <code>&nbsp;</code><code>}catch(InterruptedException ex){ </code> <code>&nbsp;</code><code>ex.printStackTrace();</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code>&nbsp; <code>&nbsp;</code><code>@Override</code> <code>&nbsp;</code><code>public void run() { </code> <code>&nbsp;</code><code>// TODO Auto-generated method stub</code> <code>&nbsp;</code><code>try{ </code> <code>&nbsp;</code><code>synchronized(this){ </code> <code>&nbsp;</code><code>while(alive){ </code> <code>&nbsp;</code><code>wait();</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}catch(InterruptedException ex){ </code> <code>&nbsp;</code><code>ex.printStackTrace();</code> <code>&nbsp;</code><code>Thread.currentThread().interrupt();</code> <code>&nbsp;</code><code>}finally{ </code> <code>&nbsp;</code><code>this.close();</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;</code><code>public static void main(String[] args) throws IOException,InterruptedException,KeeperException{ </code> <code>&nbsp;</code><code>if(args.length != 1){ </code> <code>&nbsp;</code><code>System.err.println("Usage:ClusterMonitor&lt;</code><code>Host:Port</code><code>&gt;");</code> <code>&nbsp;</code><code>System.exit(0);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>String hostPort = args[0];</code> <code>&nbsp;</code><code>new ClusterMonitor(hostPort).run();</code> <code>&nbsp;</code><code>}</code> &nbsp; <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

清单 2.ClusterClient 类

清单 3.Eclipse 运行输出

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> <code>childrenWatcher Event Received:%sWatchedEvent state:SyncConnected type:NodeChildrenChanged path:/Members</code> <code>Cluster Membership change,Members: [dweref0000000009, test100000000003, dsdawqeqw0000000008, </code> <code>test111110000000004, test22220000000005, dsda32130000000007, dsda0000000006, test10000000002]</code> </td> 
 </tr> 
 </tbody> 
</table>

我们通过 zkCli 方式对被监听的/Members 这个 ZNODE 操作，增加一个子节点，您会在 zkCli 里看到如清单 4 所示输出。

清单 4.ZKCli 创建 ZNode 子节点

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 </td> 
 <td> <code>[zk: localhost:2181(CONNECTED) 0] create -s /Members/dweref rew23rf</code> <code>Created /Members/dweref0000000009 [zk: localhost:2181(CONNECTED) 4]</code> </td> 
 </tr> 
 </tbody> 
</table>

上面的示例我们演示了如何发起对于一个 ZNODE 的监听，当该 ZNODE 被改变后，我们会触发对应的方法进行处理，这类方式可以被用在数据监听、集群状态监听等用途。

### 回调函数 ###

由于 Watcher 机制涉及到回调函数，所以我们先来介绍一下回调函数的基础知识。

打个比方，有一家旅馆提供叫醒服务，但是要求旅客自己决定叫醒的方法。可以是打客房电话，也可以是派服务员去敲门，睡得死怕耽误事的，还可以要求往自己头上浇盆水。这里，“叫醒”这个行为是旅馆提供的，相当于库函数，但是叫醒的方式是由旅客决定并告诉旅馆的，也就是回调函数。而旅客告诉旅馆怎么叫醒自己的动作，也就是把回调函数传入库函数的动作，称为登记回调函数（to register a callback function）。

乍看起来，回调似乎只是函数间的调用，但仔细一琢磨，可以发现两者之间的一个关键的不同：在回调中，我们利用某种方式，把回调函数像参数一样传入中间函数。可以这么理解，在传入一个回调函数之前，中间函数是不完整的。换句话说，程序可以在运行时，通过登记不同的回调函数，来决定、改变中间函数的行为。这就比简单的函数调用要灵活太多了。

回调实际上有两种：阻塞式回调和延迟式回调。两者的区别在于：阻塞式回调里，回调函数的调用一定发生在起始函数返回之前；而延迟式回调里，回调函数的调用有可能是在起始函数返回之后。我们来看一个简单的示例。

清单 5.Caller 类

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>public class Caller </code> <code>{ </code> <code>&nbsp;</code><code>public MyCallInterface mc; </code> <code>&nbsp;</code>&nbsp; <code>&nbsp;</code><code>public void setCallfuc(MyCallInterface mc) </code> <code>&nbsp;</code><code>{ </code> <code>&nbsp;</code><code>this.mc= mc; </code> <code>&nbsp;</code><code>} </code> <code>&nbsp;</code>&nbsp; <code>&nbsp;</code><code>public void call(){ </code> <code>&nbsp;</code><code>this.mc.method(); </code> <code>&nbsp;</code><code>} </code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

清单 6.MyCallInterface 接口

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> <code>public interface MyCallInterface { </code> <code>public void method(); </code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

清单 7.CallbackClass 类

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>public class CallbackClass implements MyCallInterface{ </code> <code>public void method() </code> <code>&nbsp;</code><code>{ </code> <code>&nbsp;</code><code>System.out.println("回调函数"); </code> <code>&nbsp;</code><code>} </code> <code>&nbsp;</code>&nbsp; <code>&nbsp;</code><code>public static void main(String args[]) </code> <code>&nbsp;</code><code>{ </code> <code>&nbsp;</code><code>Caller call = new Caller(); </code> <code>&nbsp;</code><code>call.setCallfuc(new CallbackClass()); </code> <code>&nbsp;</code><code>call.call(); </code> <code>&nbsp;</code><code>} </code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

清单 8. 运行结果

### 原理及源代码解释 ###

**实现原理**

ZooKeeper 允许客户端向服务端注册一个 Watcher 监听，当服务端的一些指定事件触发了这个 Watcher，那么就会向指定客户端发送一个事件通知来实现分布式的通知功能。

ZooKeeper 的 Watcher 机制主要包括客户端线程、客户端 WatchManager 和 ZooKeeper 服务器三部分。在具体工作流程上，简单地讲，客户端在向 ZooKeeper 服务器注册 Watcher 的同时，会将 Watcher 对象存储在客户端的 WatchManager 中。当 ZooKeeper 服务器端触发 Watcher 事件后，会向客户端发送通知，客户端线程从 WatchManager 中取出对应的 Watcher 对象来执行回调逻辑。如清单 9 所示，WatchManager 创建了一个 HashMap，这个 HashMap 被用来存放 Watcher 对象。

清单 9.WatchManager 类

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 </td> 
 <td> <code>private final HashMap&lt;</code><code>String</code><code>, HashSet&lt;Watcher&gt;&gt; watchTable =</code> <code>&nbsp;</code><code>new HashMap&lt;</code><code>String</code><code>, HashSet&lt;Watcher&gt;&gt;();</code> <code>public synchronized void addWatch(String path, Watcher watcher) { </code> <code>&nbsp;</code><code>HashSet&lt;</code><code>Watcher</code><code>&gt; list = watchTable.get(path);</code> <code>&nbsp;</code><code>if (list == null) { </code> <code>&nbsp;</code><code>// don't waste memory if there are few watches on a node</code> <code>&nbsp;</code><code>// rehash when the 4th entry is added, doubling size thereafter</code> <code>&nbsp;</code><code>// seems like a good compromise</code> <code>&nbsp;</code><code>list = new HashSet&lt;</code><code>Watcher</code><code>&gt;(4);</code> <code>&nbsp;</code><code>watchTable.put(path, list);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>list.add(watcher);</code> &nbsp; <code>&nbsp;</code><code>HashSet&lt;</code><code>String</code><code>&gt; paths = watch2Paths.get(watcher);</code> <code>&nbsp;</code><code>if (paths == null) { </code> <code>&nbsp;</code><code>// cnxns typically have many watches, so use default cap here</code> <code>&nbsp;</code><code>paths = new HashSet&lt;</code><code>String</code><code>&gt;();</code> <code>&nbsp;</code><code>watch2Paths.put(watcher, paths);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>paths.add(path);</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

整个 Watcher 注册和通知流程如图 1 所示。

图 1.Watcher 注册和通知流程图

![img001.png][]

**Watcher 接口**

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

清单 10.Watcher 接口

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> <code>public interface Watcher { </code> <code>abstract public void process(WatchedEvent event);</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

在 Watcher 接口里面，除了回调函数 process 以外，还包含 KeeperState 和 EventType 两个枚举类，分别代表了通知状态和事件类型，如图 2 所示。

图 2.Watcher通知状态和事件类型表

![img003.png][]

process 方法是 Watcher 接口中的一个回调方法，当 ZooKeeper 向客户端发送一个 Watcher 事件通知时，客户端就会对相应的 process 方法进行回调，从而实现对事件的处理。

process 方法包含 WatcherEvent 类型的参数，WatchedEvent 包含了每一个事件的三个基本属性：通知状态（KeeperState）、事件类型（EventType）和节点路径（Path），ZooKeeper 使用 WatchedEvent 对象来封装服务端事件并传递给 Watcher，从而方便回调方法 process 对服务端事件进行处理。

WatchedEvent 和 WatcherEvent 都表示的是同一个事物，都是对一个服务端事件的封装。不同的是，WatchedEvent 是一个逻辑事件，用于服务端和客户端程序执行过程中所需的逻辑对象，而 WatcherEvent 因为实现了序列化接口，因此可以用于网络传输。

服务端在线程 WatchedEvent 事件之后，会调用 getWrapper 方法将自己包装成一个可序列化的 WatcherEvent 事件，如清单 7 所示，以便通过网络传输到客户端。客户端在接收到服务端的这个事件对象后，首先会将 WatcherEvent 事件还原成一个 WatchedEvent 事件，并传递给 process 方法处理，回调方法 process 根据入参就能够解析出完整的服务端事件了。

清单 11. 可序列化的事件

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 </td> 
 <td> <code>public WatcherEvent getWrapper() { </code> <code>&nbsp;</code><code>return new WatcherEvent(eventType.getIntValue(), </code> <code>&nbsp;</code><code>keeperState.getIntValue(), </code> <code>&nbsp;</code><code>path);</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

**客户端注册 Watcher 流程**

清单 1 所示代码中采用了 ZooKeeper 构造函数来传入一个 Watcher，如代码 zk = new ZooKeeper(HostPort,2000,connectionWatcher);在这行代码里，第三个参数是连接到 ZooKeeper 服务端的 connectionWatcher 事件监听，这个 Watcher 将作为整个 ZooKeeper 会话期间的默认 Watcher，会一直被保存在客户端 ZKWatchManager 的 defaultWatcher 里面。

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

清单 12.getChildren 方法添加 watch 事件

在 ZooKeeper 中，Packet 是一个最小的通信协议单元，即数据包。Pakcet 用于进行客户端与服务端之间的网络传输，任何需要传输的对象都需要包装成一个 Packet 对象。在 ClientCnxn 中 WatchRegistration 也会被封装到 Pakcet 中，然后由 SendThread 线程调用 queuePacke 方法把 Packet 放入发送队列中等待客户端发送，这又是一个异步过程，分布式系统采用异步通信是一个普遍认同的观念。随后，SendThread 线程会通过 readResponse 方法接收来自服务端的响应，异步地调用 finishPacket 方法从 Packet 中取出对应的 Watcher 并注册到 ZKWatchManager 中去，如清单 13 所示。

清单 13.getChildren 方法添加 watch 事件

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 </td> 
 <td> <code>private void finishPacket(Packet p) { </code> <code>&nbsp;</code><code>if (p.watchRegistration != null) { </code> <code>&nbsp;</code><code>p.watchRegistration.register(p.replyHeader.getErr());</code> <code>&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;</code><code>if (p.cb == null) { </code> <code>&nbsp;</code><code>synchronized (p) { </code> <code>&nbsp;</code><code>p.finished = true;</code> <code>&nbsp;</code><code>p.notifyAll();</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>} else { </code> <code>&nbsp;</code><code>p.finished = true;</code> <code>&nbsp;</code><code>eventThread.queuePacket(p);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

除了上面介绍的方式以外，ZooKeeper 客户端也可以通过 getData、getChildren 和 exist 三个接口来向 ZooKeeper 服务器注册 Watcher，无论使用哪种方式，注册 Watcher 的工作原理是一致的。如清单 14 所示，getChildren 方法调用了 WatchManager 类的 addWatch 方法添加了 watcher 事件。

清单 14.getChildren 方法添加 watcher 事件

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 </td> 
 <td> <code>public ArrayList&lt;</code><code>String</code><code>&gt; getChildren(String path, Stat stat, </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Watcher watcher) throws KeeperException.NoNodeException { </code> <code>&nbsp;</code><code>DataNodeV1 n = nodes.get(path);</code> <code>&nbsp;</code><code>if (n == null) { </code> <code>&nbsp;</code><code>throw new KeeperException.NoNodeException();</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>synchronized (n) { </code> <code>&nbsp;</code><code>ArrayList&lt;</code><code>String</code><code>&gt; children = new ArrayList&lt;</code><code>String</code><code>&gt;();</code> <code>&nbsp;</code><code>children.addAll(n.children);</code> <code>&nbsp;</code><code>if (watcher != null) { </code> <code>&nbsp;</code><code>childWatches.addWatch(path, watcher);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>return children;</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

如清单 15 所示，现在需要从这个封装对象中再次提取出 Watcher 对象来，在 register 方法里面，客户端将 Watcher 对象转交给 ZKWatchManager，并最终保存在一个 Map 类型的数据结构 dataWatches 里面，用于将数据节点的路径和 Watcher 对象进行一一映射后管理起来。

注意，WatcherRegistation 除了 Header 和 request 两个属性被传递到了服务端，其他都没有到服务端，否则服务端就容易出现内存紧张甚至溢出的危险，因为数据量太大了。这就是 ZooKeeper 为什么适用于分布式环境的原因，它在网络中传输的是消息，而不是数据包实体。

清单 15.processRequest 代码

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>public void register(int rc) { </code> <code>&nbsp;</code><code>if (shouldAddWatch(rc)) { </code> <code>&nbsp;</code><code>Map&lt;</code><code>String</code><code>, Set&lt;Watcher&gt;&gt; watches = getWatches(rc);</code> <code>&nbsp;</code><code>synchronized(watches) { </code> <code>&nbsp;</code><code>Set&lt;</code><code>Watcher</code><code>&gt; watchers = watches.get(clientPath);</code> <code>&nbsp;</code><code>if (watchers == null) { </code> <code>&nbsp;</code><code>watchers = new HashSet&lt;</code><code>Watcher</code><code>&gt;();</code> <code>&nbsp;</code><code>watches.put(clientPath, watchers);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>watchers.add(watcher);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

**服务端处理 Watcher 流程**

如图 3 所示是服务端处理 Watcher 的一个完整序列图。

图 3. 服务端处理 Watcher 序列图

![img002.jpg][]

注意，以下所有代码均为精简版，去除了日志、判断分支，只在源码上保留了主线代码。

FinalRequestProcessor 类接收到客户端请求后，会调用 processRequest 方法进行处理，会进一步转向 ZooKeeperServer 的 processRequest 进行进一步处理，处理结由 ZKDatabase 类返回，如清单 16-18 所示。

清单 16.processRequest 代码

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 </td> 
 <td> <code>public void processRequest(Request request) { </code> <code>if (request.hdr != null) { </code> <code>&nbsp;</code><code>TxnHeader hdr = request.hdr;</code> <code>&nbsp;</code><code>Record txn = request.txn;</code> &nbsp; <code>&nbsp;</code><code>rc = zks.processTxn(hdr, txn);</code> <code>&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

清单 17.ZooKeeperServer 代码

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 </td> 
 <td> <code>public ProcessTxnResult processTxn(TxnHeader hdr, Record txn) { </code> <code>ProcessTxnResult rc;</code> <code>&nbsp;</code><code>int opCode = hdr.getType();</code> <code>&nbsp;</code><code>long sessionId = hdr.getClientId();</code> <code>&nbsp;</code><code>rc = getZKDatabase().processTxn(hdr, txn);</code> <code>&nbsp;</code><code>if (opCode == OpCode.createSession) { </code> <code>&nbsp;</code><code>if (txn instanceof CreateSessionTxn) { </code> <code>&nbsp;</code><code>CreateSessionTxn cst = (CreateSessionTxn) txn;</code> <code>&nbsp;</code><code>sessionTracker.addSession(sessionId, cst</code> <code>&nbsp;</code><code>.getTimeOut());</code> <code>&nbsp;</code><code>} else { </code> <code>&nbsp;</code><code>LOG.warn("*****&gt;&gt;&gt;&gt;&gt; Got "</code> <code>&nbsp;</code><code>+ txn.getClass() + " "</code> <code>&nbsp;</code><code>+ txn.toString());</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>} else if (opCode == OpCode.closeSession) { </code> <code>&nbsp;</code><code>sessionTracker.removeSession(sessionId);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>return rc;</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

清单 18.ZKDatabase 代码

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 </td> 
 <td> <code>public ProcessTxnResult processTxn(TxnHeader header, Record txn)</code> <code>&nbsp;</code><code>{ </code> <code>switch (header.getType()) { </code> <code>case OpCode.setData:</code> <code>&nbsp;</code><code>SetDataTxn setDataTxn = (SetDataTxn) txn;</code> <code>&nbsp;</code><code>rc.path = setDataTxn.getPath();</code> <code>&nbsp;</code><code>rc.stat = setData(setDataTxn.getPath(), setDataTxn</code> <code>&nbsp;</code><code>.getData(), setDataTxn.getVersion(), header</code> <code>&nbsp;</code><code>.getZxid(), header.getTime());</code> <code>&nbsp;</code><code>break;</code> </td> 
 </tr> 
 </tbody> 
</table>

对于注册 Watcher 请求，FinalRequestProcessor 的 ProcessRequest 方法会判断当前请求是否需要注册 Watcher，如果为 true，就会将当前的 ServerCnxn 对象和数据节点路径传入 getData 方法中去。ServerCnxn 是一个 ZooKeeper 客户端和服务器之间的连接接口，代表了一个客户端和服务器的连接，我们后面讲到的 process 回调方法，实际上也是从这里回调的，所以可以把 ServerCnxn 看作是一个 Watcher 对象。数据节点的节点路径和 ServerCnxn 最终会被存储在 WatchManager 的 watchTable 和 watch2Paths 中。

清单 19. 判断是否注册 Watcher 代码

如前所述，WatchManager 负责 Watcher 事件的触发，它是一个统称，在服务端 DataTree 会托管两个 WatchManager，分别是 dataWatches 和 childWatches，分别对应数据变更 Watcher 和子节点变更 Watcher。

清单 20.WatchManger 两个队列

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 </td> 
 <td> <code>private final HashMap&lt;</code><code>String</code><code>, HashSet&lt;Watcher&gt;&gt; watchTable =</code> <code>&nbsp;</code><code>new HashMap&lt;</code><code>String</code><code>, HashSet&lt;Watcher&gt;&gt;();</code> &nbsp; <code>&nbsp;</code><code>private final HashMap&lt;</code><code>Watcher</code><code>, HashSet&lt;String&gt;&gt; watch2Paths =</code> <code>&nbsp;</code><code>new HashMap&lt;</code><code>Watcher</code><code>, HashSet&lt;String&gt;&gt;();</code> </td> 
 </tr> 
 </tbody> 
</table>

回到主题，如清单 21 到 23 所示，当发生 Create、Delete、NodeChange（数据变更）这样的事件后，DataTree 会调用相应方法去触发 WatchManager 的 triggerWatch 方法，该方法返回 ZNODE 的信息，自此进入到回调本地 process 的序列。

清单 21.processTxn 代码

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>public ProcessTxnResult processTxn(TxnHeader header, Record txn)</code> <code>&nbsp;</code><code>{ </code> <code>&nbsp;</code><code>ProcessTxnResult rc = new ProcessTxnResult();</code> &nbsp; <code>&nbsp;</code><code>try { </code> <code>switch (header.getType()) { </code> <code>case OpCode.setData:</code> <code>&nbsp;</code><code>SetDataTxn setDataTxn = (SetDataTxn) txn;</code> <code>&nbsp;</code><code>rc.path = setDataTxn.getPath();</code> <code>&nbsp;</code><code>rc.stat = setData(setDataTxn.getPath(), setDataTxn</code> <code>&nbsp;</code><code>.getData(), setDataTxn.getVersion(), header</code> <code>&nbsp;</code><code>.getZxid(), header.getTime());</code> <code>&nbsp;</code><code>break;</code> </td> 
 </tr> 
 </tbody> 
</table>

清单 22.setData 代码

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 </td> 
 <td> <code>public Stat setData(String path, byte data[], int version, long zxid,</code> <code>&nbsp;</code><code>long time) throws KeeperException.NoNodeException { </code> <code>&nbsp;</code><code>Stat s = new Stat();</code> <code>&nbsp;</code><code>DataNodeV1 n = nodes.get(path);</code> <code>&nbsp;</code><code>if (n == null) { </code> <code>&nbsp;</code><code>throw new KeeperException.NoNodeException();</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>synchronized (n) { </code> <code>&nbsp;</code><code>n.data = data;</code> <code>&nbsp;</code><code>n.stat.setMtime(time);</code> <code>&nbsp;</code><code>n.stat.setMzxid(zxid);</code> <code>&nbsp;</code><code>n.stat.setVersion(version);</code> <code>&nbsp;</code><code>n.copyStat(s);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>dataWatches.triggerWatch(path, EventType.NodeDataChanged);</code> <code>&nbsp;</code><code>return s;</code> <code>&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

清单 23.triggerWatch 代码

从上面的代码我们可以总结出，如果想要处理一个 Watcher，需要执行的步骤如下所示：

1. 将事件类型（EventType）、通知状态（WatchedEvent）、节点路径封装成一个 WatchedEvent 对象。

2. 根据数据节点的节点路径从 watchTable 里面取出对应的 Watcher。如果没有找到 Watcher 对象，说明没有任何客户端在该数据节点上注册过 Watcher，直接退出。如果找到了 Watcher 就将其提取出来，同时会直接从 watchTable 和 watch2Paths 里删除 Watcher，即 Watcher 是一次性的，触发一次就失效了。

3. 对于需要注册 Watcher 的请求，ZooKeeper 会把请求对应的 ServerCnxn 作为一个 Watcher 存储，所以这里调用的 process 方法实质上是 ServerCnxn 的对应方法，如清单 24 所示，在请求头标记“-1”表示当前是一个通知，将 WatchedEvent 包装成 WatcherEvent 用于网络传输序列化，向客户端发送通知，真正的回调方法在客户端，就是我们清单 10 里面定义的 process() 方法。

清单 24.ServerCnxn 类代码

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 </td> 
 <td> <code>synchronized public void process(WatchedEvent event) { </code> <code>&nbsp;</code><code>ReplyHeader h = new ReplyHeader(-1, -1L, 0);</code> <code>&nbsp;</code><code>if (LOG.isTraceEnabled()) { </code> <code>&nbsp;</code><code>ZooTrace.logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK,</code> <code>&nbsp;</code><code>"Deliver event " + event + " to 0x"</code> <code>&nbsp;</code><code>+ Long.toHexString(this.sessionId)</code> <code>&nbsp;</code><code>+ " through " + this);</code> <code>&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;</code><code>// Convert WatchedEvent to a type that can be sent over the wire</code> <code>&nbsp;</code><code>WatcherEvent e = event.getWrapper();</code> &nbsp; <code>&nbsp;</code><code>sendResponse(h, e, "notification");</code> <code>&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

如清单 24 所示，客户端收到消息后，会调用 ClientCnxn 的 SendThread.readResponse 方法来进行统一处理，如清单所示。如果响应头 replyHdr 中标识的 Xid 为 02，表示是 ping，如果为-4，表示是验证包，如果是-1，表示这是一个通知类型的响应，然后进行反序列化、处理 chrootPath、还原 WatchedEvent、回调 Watcher 等步骤，其中回调 Watcher 步骤将 WacthedEvent 对象交给 EventThread 线程，在下一个轮询周期中进行 Watcher 回调。

清单 25.SendThread 线程代码

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 </td> 
 <td> <code>class SendThread extends ZooKeeperThread { </code> <code>&nbsp;</code><code>private long lastPingSentNs;</code> <code>&nbsp;</code><code>private final ClientCnxnSocket clientCnxnSocket;</code> <code>&nbsp;</code><code>private Random r = new Random(System.nanoTime()); </code> <code>&nbsp;</code><code>private boolean isFirstConnect = true;</code> &nbsp; <code>&nbsp;</code><code>void readResponse(ByteBuffer incomingBuffer) throws IOException { </code> <code>&nbsp;</code><code>ByteBufferInputStream bbis = new ByteBufferInputStream(</code> <code>&nbsp;</code><code>incomingBuffer);</code> <code>&nbsp;</code><code>BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);</code> <code>&nbsp;</code><code>ReplyHeader replyHdr = new ReplyHeader();</code> &nbsp; <code>&nbsp;</code><code>replyHdr.deserialize(bbia, "header");</code> <code>&nbsp;</code><code>if (replyHdr.getXid() == -2) { </code> </td> 
 </tr> 
 </tbody> 
</table>

如清单 25 所示，SendThread 接收到服务端的通知事件后，会通过调用 EventThread 类的 queueEvent 方法将事件传给 EventThread 线程，queueEvent 方法根据该通知事件，从 ZKWatchManager 中取出所有相关的 Watcher，如清单 26 所示。

清单 26.EventThread 线程代码

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 </td> 
 <td> <code>class EventThread extends ZooKeeperThread { </code> <code>public void queueEvent(WatchedEvent event) { </code> <code>&nbsp;</code><code>if (event.getType() == EventType.None</code> <code>&nbsp;</code><code>&amp;&amp; sessionState == event.getState()) { </code> <code>&nbsp;</code><code>return;</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>sessionState = event.getState();</code> &nbsp; <code>&nbsp;</code><code>// materialize the watchers based on the event</code> <code>&nbsp;</code><code>WatcherSetEventPair pair = new WatcherSetEventPair(</code> <code>&nbsp;</code><code>watcher.materialize(event.getState(), event.getType(),</code> <code>&nbsp;</code><code>event.getPath()),</code> <code>&nbsp;</code><code>event);</code> <code>&nbsp;</code><code>// queue the pair (watch set &amp; event) for later processing</code> <code>&nbsp;</code><code>waitingEvents.add(pair);</code> <code>&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

客户端在识别出事件类型 EventType 之后，会从相应的 Watcher 存储中删除对应的 Watcher，获取到相关的 Watcher 之后，会将其放入 waitingEvents 队列，该队列从字面上就能理解是一个待处理队列，线程的 run 方法会不断对该该队列进行处理，这就是一种异步处理思维的实现。

清单 27.ZKWatchManager 取出 Watcher

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>public Set&lt;</code><code>Watcher</code><code>&gt; materialize(Watcher.Event.KeeperState state,</code> <code>&nbsp;</code><code>Watcher.Event.EventType type,</code> <code>&nbsp;</code><code>String clientPath)</code> <code>&nbsp;</code><code>{ </code> <code>&nbsp;</code><code>Set&lt;</code><code>Watcher</code><code>&gt; result = new HashSet&lt;</code><code>Watcher</code><code>&gt;();</code> <code>case NodeCreated:</code> <code>&nbsp;</code><code>synchronized (dataWatches) { </code> <code>&nbsp;</code><code>addTo(dataWatches.remove(clientPath), result);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>synchronized (existWatches) { </code> <code>&nbsp;</code><code>addTo(existWatches.remove(clientPath), result);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>break;</code> </td> 
 </tr> 
 </tbody> 
</table>

清单 28.EventThread 线程的 run 方法

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 </td> 
 <td> <code>public void run() { </code> <code>&nbsp;</code><code>try { </code> <code>&nbsp;</code><code>isRunning = true;</code> <code>&nbsp;</code><code>while (true) { </code> <code>&nbsp;</code><code>Object event = waitingEvents.take();</code> <code>&nbsp;</code><code>if (event == eventOfDeath) { </code> <code>&nbsp;</code><code>wasKilled = true;</code> <code>&nbsp;</code><code>} else { </code> <code>&nbsp;</code><code>processEvent(event);</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>if (wasKilled)</code> <code>&nbsp;</code><code>synchronized (waitingEvents) { </code> <code>&nbsp;</code><code>if (waitingEvents.isEmpty()) { </code> <code>&nbsp;</code><code>isRunning = false;</code> <code>&nbsp;</code><code>break;</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> <code>&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

[回到顶部][Link 5]

## ZooKeeper Watcher 特性总结 ##

**1. 注册只能确保一次消费**

无论是服务端还是客户端，一旦一个 Watcher 被触发，ZooKeeper 都会将其从相应的存储中移除。因此，开发人员在 Watcher 的使用上要记住的一点是需要反复注册。这样的设计有效地减轻了服务端的压力。如果注册一个 Watcher 之后一直有效，那么针对那些更新非常频繁的节点，服务端会不断地向客户端发送事件通知，这无论对于网络还是服务端性能的影响都非常大。

**2. 客户端串行执行**

客户端 Watcher 回调的过程是一个串行同步的过程，这为我们保证了顺序，同时，需要开发人员注意的一点是，千万不要因为一个 Watcher 的处理逻辑影响了整个客户端的 Watcher 回调。

**3. 轻量级设计**

WatchedEvent 是 ZooKeeper 整个 Watcher 通知机制的最小通知单元，这个数据结构中只包含三部分的内容：通知状态、事件类型和节点路径。也就是说，Watcher 通知非常简单，只会告诉客户端发生了事件，而不会说明事件的具体内容。例如针对 NodeDataChanged 事件，ZooKeeper 的 Watcher 只会通知客户指定数据节点的数据内容发生了变更，而对于原始数据以及变更后的新数据都无法从这个事件中直接获取到，而是需要客户端主动重新去获取数据，这也是 ZooKeeper 的 Watcher 机制的一个非常重要的特性。另外，客户端向服务端注册 Watcher 的时候，并不会把客户端真实的 Watcher 对象传递到服务端，仅仅只是在客户端请求中使用 boolean 类型属性进行了标记，同时服务端也仅仅只是保存了当前连接的 ServerCnxn 对象。这样轻量级的 Watcher 机制设计，在网络开销和服务端内存开销上都是非常廉价的。

[回到顶部][Link 5]

## 结束语 ##

本文首先介绍了一个简单的监听示例代码，通过监听 ZNode 的变化，触发回调函数来实现触发后的业务处理，接下来简单介绍了一点回调函数的基本知识，然后我们开始讨论 Watcher 机制的实现原理，从 Watcher 接口开始聊，引申出 WatcherEvent 类型，再到添加 watcher 事件以及回调函数基本原理介绍，最后对 Watcher 机制的设计原理进行了三点总结。

[zk_watcher]: https://www.cnblogs.com/fanguangdexiaoyuer/p/7077520.html
[ZooKeeper Watcher]: https://www.cnblogs.com/fanguangdexiaoyuer/p/7077520.html#_label0
[Link 1]: https://www.cnblogs.com/fanguangdexiaoyuer/p/7077520.html#_label0_0
[Link 2]: https://www.cnblogs.com/fanguangdexiaoyuer/p/7077520.html#_label0_1
[Link 3]: https://www.cnblogs.com/fanguangdexiaoyuer/p/7077520.html#_label0_2
[ZooKeeper Watcher 1]: https://www.cnblogs.com/fanguangdexiaoyuer/p/7077520.html#_label1
[Link 4]: https://www.cnblogs.com/fanguangdexiaoyuer/p/7077520.html#_label2
[Link 5]: https://www.cnblogs.com/fanguangdexiaoyuer/p/7077520.html#_labelTop
[img001.png]: /images/20220216/7ad50563a06b4cfb8a38a703c5106cd5.png
[img003.png]: /images/20220216/c2503aeab9da49f2b592945d80a0e679.png
[img002.jpg]: /images/20220216/10991b603fec45e1a7837d2a8997f391.png