【ZooKeeper】Watcher 原理

ゝ一纸荒年。 2023-01-18 09:13 93阅读 0赞

# Watcher ----- 数据变更的通知 #

ZooKeeper 提供了分布式数据的发布/订阅功能。一个典 型的发布/订阅模型系统定义了一种一对多的订阅关系，能够让多个订阅者同时监听某一 个主题对象，当这个主题对象自身状态变化时，会通知所有订阅者，使它们能够做出相 应的处理。在 ZooKeeper中，引入了 Watcher 机制来实现这种分布式的通知功能。 ZooKeeper 允许客户端向服务端注册一个 Watcher 监听，当服务端的一些指定事件触发 了这个 Watcher, 那么就会向指定客户端发送一个事件通知来实现分布式的通知功能。 整个 Watcher 注册与通知过程如图所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70]

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

## **Watcher 接口** ##

在 ZooKeeper 中，接口类 Watcher 用于表示一个标准的事件处理器，其定义了事件通 知相关的逻辑，包含 Keeperstate 和 EventType 两个枚举类，分别代表了通知状态 和事件类型，同时定义了事件的回调方法： process(WatchedEvent event)„

### **Watcher 事件** ###

同一个事件类型在不同的通知状态中代表的含义有所不同  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70 1]

表中列举了 ZooKeeper 中最常见的几个通知状态和事件类型。其中，针对 NodeDataChanged事件，此处说的变更包括节点的数据内容 和数据的版本号 dataVersion 。因此，即使使用相同的数据内容来更新，还是会触发 这个事件通知，因为对于 ZooKeeper 来说，无论数据内容是否变更，一旦有客户端调用 了数据更新的接口，且更新成功，就会更新dataVersion 值。

NodeChildrenChanged 事件会在数据节点的子节点列表发生变更的时候被触发，这 里说的子节点列表变化特指子节点个数和组成情况的变更，即新增子节点或删除子节点， 而子节点内容的变化是不会触发这个事件的。

对于 AuthFailed 这个事件，需要注意的地方是，它的触发条件并不是简简单单因为 当前客户端会话没有权限，而是授权失败。我们首先通过两个例子来看看

// 使用正确的 Scheme 进行授权
    zkClient = new ZooKeeper( SERVERLIST, 3000, new Sample_AuthFailedl());
    zkClient.addAuthInfo(  "digest", "taokeepen:true".getBytes());
    zkClient .create( "/zk-book", "".getBytes(), ads, CreateMode.EPHEMERAL );
    
    zkClient_error = new ZooKeeper( SERVERLIST, 3000, new Sample_AuthFailedl());
    zkClient_error.addAuthlnfo(  "digest","taokeeper:error".getBytes());
    zkClient_error.getData( "/zk-book", true, null );

// 使用错误的 Scheme 进行授权
    zkClient = new ZooKeeper( SERVERLIST, 3000, new Sample_AuthFailed2());
    zkClient.addAuthlnfo(  "digest", "taokeeper:true".getBytes());
    zkClient.create( "/zk-book", "".getBytes(), acts, CreateMode.EPHEMERAL );
    
    zkClient_error = new ZooKeeper( SERVERLIST, 3000, new Sample_AuthFailed2());
    zkClient_error.addAuthlnfo(  "digest2", "taokeeper:error" .getBytes());
    zkCtient_error.getData( "/zk-book", true, null );

上面两个示例程序都创建了一个受到权限控制的数据节点，然后使用了不同的权限 Scheme 进行权限检查。在第一个示例程序中，使用了正确的权限 Scheme：digest； 而第二个示例程序中使用了错误的 Scheme： digest2。另外，无论哪个程序，都使用 了错误的 Auth ：taokeeper:error, 因此在运行第一个程序的时候，会抛出 NoAuthException 异常，而第二个程序运行后，抛出的是AuthFailedException 异常，同时，会收到对应的 Watcher 事件通知： (AuthFailed, None) 。

### **回调方法 process()** ###

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

`abstract public void process(WatchedEvent event);`

这个回调方法的定义非常简单，我们重点看下方法的参数定义： WatchedEvent 。  
WatchedEvent 包含了每一个事件的三个基本属性：通知状态 (keeperstate), 事件类型 (eventType) 和节点路径 (path), 其数据结构如图所示。 ZooKeeper 使 用 WatchedEvent 对象来封装服务端事件并传递给 Watcher, 从而方便回调方法 process 对服务端事件进行处理。  
![在这里插入图片描述][20210430153412322.png]

提到 WatchedEvent, 不得不讲下 WatcherEvent 实体。笼统地讲，两者表示的是同 一个事物，都是对一个服务端事件的封装。不同的是， WatchedEvent 是一个逻辑事 件，用于服务端和客户端程序执行过程中所需的逻辑对象，而 WatcherEvent 因为实 现了序列化接口，因此可以用于网络传输，其数据结构如图所示。  
![在这里插入图片描述][20210430153913762.png]  
服务端在生成 WatchedEvent 事件之后，会调用 getWrapper 方法将自己包装成一个 可序列化的WatcherEvent 事件，以便通过网络传输到客户端。客户端在接收到服务 端的这个事件对象后，首先会将 WatcherEvent 事件还原成一个 WatchedEvent 事 件，并传递给 process 方法处理，回调方法process 根据入参就能够解析出完整的 服务端事件了。

需要注意的一点是，无论是 WatchedEvent 还是 WatcherEvent, 其对 ZooKeeper 服 务端事件的封装都是极其简单的。举个例子来说，当 /zk-book 这个节点的数据发生变更 时，服务端会发送给客户端一个 “ZNode 数据内容变更”事件，客户端只能够接收到如 下信息：  
![在这里插入图片描述][20210430154006246.png]  
从上面展示的信息中，我们可以看到，客户端无法直接从该事件中获取到对应数据节点 的原始数据内容以及变更后的新数据内容，而是需要客户端再次主动去重新获取数 据——这也是ZooKeeper Watcher 机制的一个非常重要的特性。

# 工作原理 #

ZooKeeper 的 Watcher 机制，总的来说可以概括为以下三个过程：客户端注册 Watcher, 服务端处理 Watcher 和客户端回调 Watcher, 其内部各组件之间的关系如图所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70 2]

## **客户端注册 Watcher** ##

我们提到在创建一个 ZooKeeper 客户端对象实例时，可以向构造方法中 传入一个默认的 Watcher：

`public ZooKeeper(String connectstring, int sessionTimeout, Watcher watcher);`

这个 Watcher 将作为整个 ZooKeeper 会话期间的默认 Watcher, 会一直被保存在客户端ZKWatchManager 的 defaultWatcher 中。另外， ZooKeeper 客户端也可以通过 getData, getchildren和 exist 三个接口来向 ZooKeeper 服务器注册 Watcher, 无 论使用哪种方式，注册 Watcher 的工作原理都是一致的，这里我们以 getData 这个接 口为例来说明。 getData 接口用于获取指定节点的数据内容，主要有两个方法：

public byte[] getData(String path, boolean watch, Stat stat)
    public byte[] getData(final String path. Watcher watcher, Stat stat)

在这两个接口上都可以进行 Watcher 的注册，第一个接口通过一个 boolean 参数来标识是否使用上文中提到的默认 Watcher 来进行注册，具体的注册逻辑和第二个接口是一致 的。在向 getData 接口注册 Watcher 后，客户端首先会对当前客户端请求 request 进行标记, 将其设置为“使用 watcher 监听”，同时会封装一个 Watcher 的注册信息 Watch Registration 对象，用于暂时保存数据节点的路径和 Watcher 的对应关系，具体的逻辑代码如下：

public Stat getData(final String path, Watcher watcher, Stat stat){ 
    	...
    	WatchRegistration web = null;
    	if (watcher != null) { 
    		web = new DataWatchRegistration(watcher, clientPath);
    	}
    	...
    	request.setWatch(watcher != null);
    	ReplyHeader r = cnxn.submitRequest(h, request, response, web);
    	...
    }

在 ZooKeeper 中， Packet 可以被看作一个最小的通信协议单元，用于进行客户端与服 务端之间的网络传输，任何需要传输的对象都需要包装成一个 Packet 对象。因此，在 ClientCnxn 中WatchRegistration 又会被封装到 Packet 中去，然后放入发送队 列中等待客户端发送：

Packet queuePacket(RequestHeader h, ReplyHeader  r, Record request,
    						Record response, AsyncCallback cb, String clientPath,
    						String serverPath, Object ctx, WatchRegist ration WatchRegist ration){  
    	Packet packet = null;
    	....
    	synchronized (outgoingQueue) (
    		packet = new Packet(h,  v, request, response, WatchRegistration);
    		...
    		outgoingQueue.add(packet);
    		...
    	}

随后， ZooKeeper 客户端就会向服务端发送这个请求，同时等待请求的返回。完成请求发送 后，会由客户端 SendThread 线程的readResponse 方法负责接收来自服务端的响应，并调用 finishpacket 方法会从 Packet 中取出对应的 Watcher 并注册到ZKWatchManager 中去；

private void finishPacket(Packet p) { 
    	if (p.watchRegistration != null) { 
    		p.watchRegistration.register(p.replyHeader.getErr());  
    	}
    	...
    }

从上面的内容中，我们已经了解到客户端已经将 Watcher 暂时封装在了 WatchRegist ration 对象中，现在就需要从这个封装对象中再次提取出 Watcher 来：

protected Map<String, Set<Watcher» getWatches(int re) {  
    	return watchManager.dataWatches;
    }
    public void register(int re) { 
    	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);
    			)
    			watchers.add(watcher);
    		}
    	}
    }

在 register 方法中，客户端会将之前暂时保存的 Watcher 对象转交给 ZKWatchManager, 并最终保存到 dataWatches 中去。 ZKWatchManager. dataWatches 是一个 Map<String, Set<Watcher>> 类型的数据结构，用于将数据节点的路径和 Watcher 对象进 行一一映射后管理起来。整个客户端Watcher 的注册流程如图所示。  
![在这里插入图片描述][20210430213213406.png]

但 同时我们可以发现，极端情况下，客户端每调用一次 getData 接口，就会注册上 一个 Watcher, 那么这些 Watcher 实体都会随着客户端请求被发送到服务端去吗？答案是否定的。如果客户端注册的所有 Watcher 都被传递到服务端的话，那么服务端肯定会 出现内存紧张或其他性能问题了，幸运的是，在 ZooKeeper 的设计中充分考虑到了这个问题。在上面的流程中，我们提到把 WatchRegistration 封装到了 Packet 对象中去，但事实上，在底层实际的网络传输序列化过程中，并没有将 WatchRegistration 对象完全地序 列化到底层字节数组中去。为了证实这一点，我们可以看下 Packet 内部的序列化过程：

public void createBB() {  
    	try { 
    		ByteArrayOutputStream baos = new ByteArrayOutputStream();
    		BinaryoutputArchive boa = BinaryOutputArchive.getArchive(baos); 
    		boa.writelnt(-1,"ten"); // We'll fill this in later 
    		if (requestHeader != null){  
    			requestHeader.serialize(boa, "header");
    		}
    		if (request instanceof ConnectRequest) {  request.serialize(boa, "connect"); 
    			// append "am-I-allowed-to-be-readonly" flag 
    			boa.writeBool(readonly, "readonly");
    		} else if (request != null) {  
    			request.serialize(boa, "request");
    		}
    	}
    	...
    }

从上面的代码片段中，我们可以看到，在 Packet. createBB 方法中， ZooKeeper 只会将requestHeader 和 request 两个属性进行序列化，也就是说，尽管 WatchRegistration 被封装在了Packet 中，但是并没有被序列化到底层字节数组 中去，因此也就不会进行网络传输了。

## **服务端处理 Watcher** ##

上面主要讲解了客户端注册 Watcher 的过程，并且已经了解了最终客户端并不会将 Watcher 对象真正传递到服务端。那么，服务端究竟是如何完成客户端的 Watcher 注册， 又是如何来处理这个Watcher 的呢？

### ServerCnxn 存储 ###

我们首先来看下服务端接收 Watcher 并将其存储起来的过程，如图所示是 ZooKeeper 服务端处理 Watcher 的序列图。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70 3]

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

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

从 getData 请求的处理逻辑中,我们可以看到,当 getDataRequest. getWatch () 为 true 的时候，ZooKeeper 就认为当前客户端请求需要进行 Watcher 注册，于是就会将当前 的 ServerCnxn 对象和数据节点路径传入 getData 方法中去。那么为什么要传入 ServerCnxn 呢？

*  ServerCnxn 是一个ZooKeeper 客户端和服务器之间的连接接口，代表 了一个客户端和服务器的连接。 ServerCnxn 接  
    口的默认实现是 NIOServerCnxn, 同时 从 3.4.0 版本开始，引入了基于 Netty 的实现：NettyServerCnxn 。
 *  无论采用哪种实现方 式，都实现了 Watcher 的 process 接口，因此我们可以把ServerCnxn 看作是一个 Watcher 对象。数据节点的节点路径和 ServerCnxn 最终会被存储在 WatchManager 的 watchTable 和 watch2Paths 中。

WatchManager 是 ZooKeeper 服务端 Watcher 的管理者，其内部管理的 watchTable 和watch2Paths 两个存储结构，分别从两个维度对 Watcher 进行存储。

*  watchTable 是从数据节点路径的粒度来托管 Watcher 。
 *  watch2Paths 是从 Watcher 的粒度来控制事件触发需要触发的数据节点。

同时， WatchManager 还负责 Watcher 事件的触发，并移除那些已经被触发的 Watcher 。注意， WatchManager 是一个统称，在服务端， DataTree 中会托管两个 WatchManager, 分别是dataWatches 和 childWatches, 分别对应数据变更 Watcher 和子节点变更 Watcher 。 在本例中，因为是 getData 接口，因此最终会被存储在 dataWatches 中，其数据结构如图所示。  
![在这里插入图片描述][2021043021463287.png]

### Watcher 触发 ###

在上面的讲解中，我们了解了对于标记了 Watcher 注册的请求， ZooKeeper 会将其 对应的ServerCnxn 存储到 WatchManager 中，下面我们来看看服务端是如何触 发 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;
    	synchronized (n) (
    		lastdata = n.data;
    		n.data = data;
    		n.stat.setMtime(time);
    		n.stat.setMzxid(zxid);
    		n.stat.setVersion(version);
    		n.copyStat(s);
    	}
    	...
    	dataWatches.triggerWatch(path, EventType.NodeDataChanged); 
    	return s;
    }

在对指定节点进行数据更新后，通过调用 WatchManager 的 triggerWatch 方 法来触发相关的事件：

public Set<Watcher> triggerWatch(String path, EventType type) {  
    	return triggerWatch(path, type, null);
    }
    public Set<Watcher> triggerWatch (String path, EventType type, Set<Watcher> supress) { 
    	WatchedEvent e = new WatchedEvent(type, KeeperState.SyncConnected, path);
    	HashSet<Watcher> watchers;
    	synchronized (this) {  watchers = watchTable.remove(path);
    		// 如果不存在 Watcher, 直接返回
    		for (Watcher w : watchers) { 
    			HashSet<String> paths = watch2Paths.get(w); 
    			if (paths != null){  
    				paths.remove(path)；
    			}
    		}
    	}
    	for (Watcher w : watchers) { 
    		if (supress != null && supress.contains(w)) {  
    			continue;
    		}
    		w.process(e);
    	}
    	return watchers;
    }

无论是 dataWatches 还是 childWatches 管理器， Watcher 的触发逻辑都是一 致的，基本步骤如下。

1.  封装 WatchedEvent 。  
    首先将通知状态 (KeeperState) 、事件类型 (EventType) 以及节点路径 (Path) 封装成一个WatchedEvent 对象。
2.  查询 Watcher 。  
    根据数据节点的节点路径从 watchTable 中取出对应的 Watcher 。如果没有 找到 Watcher,说明没有任何客户端在该数据节点上注册过 Watcher, 直接退 出。而如果找到了这个Watcher, 会将其提取出来，同时会直接从 watchTable 和 watch2Paths 中将其删除一从这里我们也可以看出， Watcher 在服务端 是一次性的，即触发一次就失效了。
3.  调用 process 方法来触发 Watcher 。  
    在这一步中，会逐个依次地调用从步骤 2 中找出的所有 Watcher 的 process 方法。那么这里的 process 方法究竟做了些什么呢？在上文中我们已经提 到，对于需要注册 Watcher 的请求， ZooKeeper 会把当前请求对应的 ServerCnxn 作为一个 Watcher 进行存储，因此，这里调用的 process 方 法，事实上就是 ServerCnxn 的对应方法：

public class NIOServerCnxn extends ServerCnxn { 
    	synchronized public void process(WatchedEvent event) { 
    		ReplyHeader h = new ReplyHeader(-1, -IL, 0);
    		// Convert WatchedEvent to a type that can be sent over the wire WatcherEvent
    		e = event.getWrapper();
    		sendResponse(h, e, "notification");
    	}
    ...

从上面的代码片段中，我们可以看出在 process 方法中，主要逻辑如下。

*  在请求头中标记 “-1”， 表明当前是一个通知。
 *  将 WatchedEvent 包装成 WatcherEvent, 以便进行网络传输序列化。
 *  向客户端发送该通知。

从以上几个步骤中可以看到’ ServerCnxn 的 process 方法中的逻辑非常简单，本 质上并不是处理客户端 Watcher 真正的业务逻辑，而是借助当前客户端连接的 ServerCnxn 对象来实现对客户端的 WatchedEvent 传递，真正的客户端 Watcher 回调与业务逻辑执行都在客户端。

## **客户端回调 Watcher** ##

上面我们已经讲解了服务端是如何进行 Watcher 触发的，并且知道了最终服务端会通过 使用 ServerCnxn 对应的 TCP 连接来向客户端发送一个 WatcherEvent 事件，下面 我们来看看客户端是如何处理这个事件的。

### SendThread 接收事件通知 ###

首先我们来看下 ZooKeeper 客户端是如何接收这个客户端事件通知的：

class SendThread extends Thread { 
        void readResponse(ByteBuffer incomingBuffer) throws lOException { 
            ...
            if (replyHdr.getXid() == -1) { 
            // -1 means notification
            ...
            WatcherEvent event = new WatcherEvent();
            event.deserialize(bbia, "response");
            // convert from a server path to a client path
            if (chrootPath != null) (
                String serverPath = event.getPath();
                if (serverPath. compa reTo( ch rootPath )=0)
                	event.setPath("/");
                else if (serverPath.length() > chrootPath.length())
                	event.setPath(serverPath.substring(chrootPath.length()));
                ...
            }
            WatchedEvent we = new WatchedEvent(event);
            ...
            eventThread.queueEvent( we );
            return;
        }
        ...

对 于一 个来 自服 务端 的响 应， 客户 端都 是由 SendThread. readResponse (ByteBuffer incomingBuffer) 方法来统一进行处理的，如果响应头 replyHdr 中标识了 XID 为 T,表明这是一个通知类型的响应，对其的处理大体 上分为以下 4 个主要步骤。

1.  反序列化。  
    ZooKeeper 客户端接到请求后，首先会将字节流转换成 WatcherEvent 对象。
2.  处理 chrootPath 。  
    如果客户端设置了 chrootPath 属性，那么需要对服务端传过来的完整的节 点路径进行  
    chrootPath 处理，生成客户端的一个相对节点路径。例如客户 端设置了 chrootPath 为 /aPP1, 那么针对服务端传过来的响应包含的节点 路径为/app1/locks，经过 chrootPath  
    处理后，就会变成一个相对路径: /locks。
3.  还原 WatchedEvent 。  
    process 接口的参数定义是 WatchedEvent，因此这里需要将 WatcherEvent 对象转换成 Watched Event
4.  回调 Watcher 。  
    最后将 WatchedEvent 对象交给 EventThread 线程，在下一个轮询周期 中进行 Watcher 回调。

### EventThread 处理事件通知 ###

在上面内容中我们讲到，服务端的 Watcher 事件通知，最终交给了 EventThread 线程来处理，现在我们就来看看 EventThread 的一些核心逻辑。 EventThread 线程是 ZooKeeper 客户端中专门用来处理服务端通知事件的线程，其数据结构如 图所示。

![在这里插入图片描述][20210502150558217.png]

在上文中，我们讲到 SendThread 接收到服务端的通知事件后，会通过调用 EventThread .queueEvent 方法将事件传给 EventThread 线程，其逻辑如下 ；

public void queueEvent(WatchedEvent event) { 
        if (event.getType() == EventType.None
                && sessionstate = event.getState()) {  
            return;
        }
        sessionstate = event.getState();
        // materialize the watchers based on the event
        WatcherSetEvent Pair pair = new WatcherSetEventPair( 
                    watcher.materialize(event.getState(), event.getType(),
                                event.getPath()), event);
        // queue the pair (watch set & event) for later processing 
        waitingEvents.add(pair);
    }

queueEvent 方法首先会根据该通知事件，从 ZKWatchManager 中取出所有相关 的 Watcher：

public Set<Watcher> materialize(Watcher.Event.Keeperstate state,
        Watcher.Event.EventType type. String clientPath){ 
        Set<Watcher> result = new HashSet<Watcher>();
        switch (type) { 
                ...
            case NodeDataChanged:
            case NodeCreated:
                synchronized (dataWatches) {  
                    addTo(dataWatches.remove(clientPath), result);
                }
                synchronized (existWatches) {  
                    addTo(existWatches.remove(clientPath),result);
                }
            break;
            ...
        return result;
    }
    
    final private void addTo(Set<Watcher> from, Set<Watcher> to) {  
    	if (from != null) { 
    		to.addAll(from);
        }
    }

客户端在识别出事件类型 EventType 后，会从相应的 Watcher 存储(即 dataWatches 、existWatches 或 childWatches 中的一个或多个，本例中就 是从 dataWatches 和 existWatches 两个存储中获取)中去除对应的 Watcher 。  
注意，此处使用的是 remove 接口，因此也表明了客户端的 Watcher 机制同样也是 一次性的，即一旦被触发后，该 Watcher 就失效了。

获取到相关的所有 Watcher 之后，会将其放入 waitingEvents 这个队列中去。  
WaitingEvents 是一个待处理 Watcher 的队列， EventTh read 的 run 方法会不 断对该队列进行处理：

public void run() {  
        try { 
            isRunning = true;
            while (true) { 
            	Object event = waitingEvents.take(); 
                if (event = eventOfDeath) {  
                    wasKilled = true;
            	} else {  
                    processEvent(event);
            	} 
                //......
            }
         ...
    }
    private void processEvent(Object event) {  
        try { 
            if (event instanceof WatcherSetEventPair) {  
            // each watcher will process the event
            WatcherSetEventPair pair = (WatcherSetEventPair) event; 
            for (Watcher watcher : pair.watchers) {  
                try (
                	watcher.process(pair.event);
                } catch (Throwable t) { 
         ...

从上面的代码片段中我们可以看出， EventThread 线程每次都会从 waiting Events 队列中取出一个 Watcher, 并进行串行同步处理。注意，此处 processEvent 方法中的 Watcher 才是之前客户端真正注册的 Watcher, 调用其 process 方法就可以实现 Watcher 的回调了。

# **Watcher 特性总结** #

到目前为止，相信读者已经了解了 ZooKeeper 中 Watcher 机制的相关接口定义以及 Watcher 的各类事件。同时，我们以 ZooKeeper 节点的数据内容获取接口为例，从 ZooKeeper 客户端进行Watcher 注册、服务端处理 Watcher 以及客户端回调 Watcher 三方 面分阶段讲解了 ZooKeeper 的Watcher 工作机制。通过上面内容的讲解，我们不难发现 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 机制设计，在网络开销和服务端内存开销上都是非常廉价的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70]: /images/20221021/bb60ba6ec4a84a03a103a21a36d9968f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70 1]: /images/20221021/f05a3ed1b937416c8fc3ee6656cdd141.png
[20210430153412322.png]: /images/20221021/beb0b72e49e74bdeac3f0ae66fd3cc5c.png
[20210430153913762.png]: /images/20221021/42dde2aa132a40edb06cb588b4f11591.png
[20210430154006246.png]: /images/20221021/e90c77cd351a4682b0a5e0ce42e5da6b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70 2]: /images/20221021/bd4cc64e658c48e09bf9063b500e389f.png
[20210430213213406.png]: /images/20221021/586557e17c8f400a843aacb28e96ed22.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNDYwNw_size_16_color_FFFFFF_t_70 3]: /images/20221021/f28019b74b894ab3856b0a0e722db287.png
[2021043021463287.png]: /images/20221021/0750c3b8e1174deca0dc6392f99cdd04.png
[20210502150558217.png]: /images/20221021/768e78964e694afdaaa832277d82d7c6.png