zookeeper之事件触发

前面这么长的说明，只是为了清洗的说明事件的注册流程，最终的触发，还得需要通过事务型操作来完成。

在我们最开始的案例中，通过如下代码去完成了事件的触发。

zookeeper.setData(“/mic”, “1”.getByte(),-1) ; //修改节点的值触发监听

前面的客户端和服务端对接的流程就不再重复讲解了，交互流程是一样的，唯一的差别在于事件触发了。

服务端的事件响应 DataTree.setData()

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);
		}
		// now update if the path is in a quota subtree.
		String lastPrefix = getMaxPrefixWithQuota(path);
		if (lastPrefix != null) {
			this.updateBytes(lastPrefix, (data == null ? 0 : data.length) - (lastdata == null ? 0 : lastdata.length));
		}
		dataWatches.triggerWatch(path, EventType.NodeDataChanged); // 触发对应节点的NodeDataChanged事件
		return s;
	}

WatcherManager. triggerWatch

 

Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {
		WatchedEvent e = new WatchedEvent(type, KeeperState.SyncConnected, path); // 根据事件类型、连接状态、节点路径创建
																					// WatchedEvent
		HashSet<Watcher> watchers;
		synchronized (this) {
			watchers = watchTable.remove(path); // 从 watcher 表中移除 path，并返回其对应的
												// watcher 集合
			if (watchers == null || watchers.isEmpty()) {
				if (LOG.isTraceEnabled()) {
					ZooTrace.logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK, "No watchers for " + path);
				}
				return null;
			}
			for (Watcher w : watchers) { // 遍历 watcher 集合
				HashSet<String> paths = watch2Paths.get(w); // 根据 watcher 从
															// watcher 表中取出路径集合
				if (paths != null) {
					paths.remove(path); // 移除路径
				}
			}
		}
		for (Watcher w : watchers) { // 遍历 watcher 集合
			if (supress != null && supress.contains(w)) {
				continue;
			}
			w.process(e); // OK，重点又来了，w.process 是做什么呢？
		}
		return watchers;
	}

 w.process(e);

还记得我们在服务端绑定事件的时候，watcher 绑定是是什么？是 ServerCnxn，所以 w.process(e)，其实调用的应该是 ServerCnxn 的 process 方法。而servercnxn 又是一个抽象方法，有两个实现类，分别是：NIOServerCnxn 和NIOServerCnxn。那接下来我们扒开 NIOServerCnxn 这个类的 process 方法看看究竟。

public void process(WatchedEvent event) {
		ReplyHeader h = new ReplyHeader(-1, -1L, 0);
		if (LOG.isTraceEnabled()) {
			ZooTrace.logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK,
					"Deliver event " + event + " to 0x" + Long.toHexString(this.sessionId) + " through " + this);
		}
		// Convert WatchedEvent to a type that can be sent over the wire
		WatcherEvent e = event.getWrapper();
		try {
			sendResponse(h, e, "notification"); // look， 这个地方发送了一个事件，事件对象为
												// WatcherEvent。完美
		} catch (IOException e1) {
			if (LOG.isDebugEnabled()) {
				LOG.debug("Problem sending to " + getRemoteSocketAddress(), e1);
			}
			close();
		}
	}

那接下里，客户端会收到这个 response，触发 SendThread.readResponse 方法。

客户端处理事件响应

SendThread.readResponse

这块代码上面已经贴过了，所以我们只挑选当前流程的代码进行讲解，按照前面我们将到过的，notifacation 通知消息的 xid 为-1，意味着~直接找到-1 的判断进行分析。

void readResponse(ByteBuffer incomingBuffer) throws IOException {
		ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);
		BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
		ReplyHeader replyHdr = new ReplyHeader();
		replyHdr.deserialize(bbia, "header");
		if (replyHdr.getXid() == -2) { // ?
			// -2 is the xid for pings
			if (LOG.isDebugEnabled()) {
				LOG.debug("Got ping response for sessionid: 0x" + Long.toHexString(sessionId) + " after "
						+ ((System.nanoTime() - lastPingSentNs) / 1000000) + "ms");
			}
			return;
		}
		if (replyHdr.getXid() == -4) {
			// -4 is the xid for AuthPacket
			if (replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
				state = States.AUTH_FAILED;
				eventThread.queueEvent(
						new WatchedEvent(Watcher.Event.EventType.None, Watcher.Event.KeeperState.AuthFailed, null));
			}
			if (LOG.isDebugEnabled()) {
				LOG.debug("Got auth sessionid:0x" + Long.toHexString(sessionId));
			}
			return;
		}
		if (replyHdr.getXid() == -1) {
			// -1 means notification
			if (LOG.isDebugEnabled()) {
				LOG.debug("Got notification sessionid:0x" + Long.toHexString(sessionId));
			}
			WatcherEvent event = new WatcherEvent();
			event.deserialize(bbia, "response"); // 这个地方，是反序列化服务端的 WatcherEvent
													// 事件。
			// convert from a server path to a client path
			if (chrootPath != null) {
				String serverPath = event.getPath();
				if (serverPath.compareTo(chrootPath) == 0)
					event.setPath("/");
				else if (serverPath.length() > chrootPath.length())
					event.setPath(serverPath.substring(chrootPath.length()));
				else {
					LOG.warn(
							"Got server path " + event.getPath() + " which is too short for chroot path " + chrootPath);
				}
			}
			WatchedEvent we = new WatchedEvent(event); // 组装 watchedEvent 对象。
			if (LOG.isDebugEnabled()) {
				LOG.debug("Got " + we + " for sessionid 0x" + Long.toHexString(sessionId));
			}
			eventThread.queueEvent(we); // 通过 eventTherad 进行事件处理
			return;
		}
		// If SASL authentication is currently in progress, construct and
		// send a response packet immediately, rather than queuing a
		// response as with other packets.
		if (tunnelAuthInProgress()) {
			GetSASLRequest request = new GetSASLRequest();
			request.deserialize(bbia, "token");
			zooKeeperSaslClient.respondToServer(request.getToken(), ClientCnxn.this);
			return;
		}
		Packet packet;
		synchronized (pendingQueue) {
			if (pendingQueue.size() == 0) {
				throw new IOException("Nothing in the queue, but got " + replyHdr.getXid());
			}
			packet = pendingQueue.remove();
		}
		/*
		 * Since requests are processed in order, we better get a r esponse to
		 * the first request!
		 */
		try {
			if (packet.requestHeader.getXid() != replyHdr.getXid()) {
				packet.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());
				throw new IOException("Xid out of order. Got Xid " + replyHdr.getXid() + " with err "
						+ +replyHdr.getErr() + " expected Xid " + packet.requestHeader.getXid()
						+ " for a packet with details: " + packet);
			}
			packet.replyHeader.setXid(replyHdr.getXid());
			packet.replyHeader.setErr(replyHdr.getErr());
			packet.replyHeader.setZxid(replyHdr.getZxid());
			if (replyHdr.getZxid() > 0) {
				lastZxid = replyHdr.getZxid();
			}
			if (packet.response != null && replyHdr.getErr() == 0) {
				packet.response.deserialize(bbia, "response");
			}
			if (LOG.isDebugEnabled()) {
				LOG.debug("Reading reply sessionid:0x" + Long.toHexString(sessionId) + ", packet::" + packet);
			}
		} finally {
			finishPacket(packet);
		}
	}

eventThread.queueEvent

SendThread 接收到服务端的通知事件后，会通过调用 EventThread 类的queueEvent 方法将事件传给 EventThread 线程，queueEvent 方法根据该通知事件，从 ZKWatchManager 中取出所有相关的 Watcher，如果获取到相应的 Watcher，就会让 Watcher 移除失效。

private void queueEvent(WatchedEvent event, Set<Watcher> materializedWatchers) {
		if (event.getType() == EventType.None && sessionState == event.getState()) { // 判断类型
			return;
		}
		sessionState = event.getState();
		final Set<Watcher> watchers;
		if (materializedWatchers == null) {
			// materialize the watchers based on the event
			watchers = watcher.materialize(event.getState(), event.getType(), event.getPath());
		} else {
			watchers = new HashSet<Watcher>();
			watchers.addAll(materializedWatchers);
		}
		// 封装 WatcherSetEventPair 对象，添加到 waitngEvents 队列中
		WatcherSetEventPair pair = new WatcherSetEventPair(watchers, event);
		// queue the pair (watch set & event) for later processing
		waitingEvents.add(pair);
	}

 

Meterialize 方法

通过 dataWatches 或者 existWatches 或者 childWatches 的 remove 取出对应的watch，表明客户端 watch 也是注册一次就移除同时需要根据 keeperState、eventType 和 path 返回应该被通知的 Watcher 集合。

public Set<Watcher> materialize(Watcher.Event.KeeperState state, Watcher.Event.EventType type, String clientPath) {
		Set<Watcher> result = new HashSet<Watcher>();
		switch (type) {
		case None:
			result.add(defaultWatcher);
			boolean clear = disableAutoWatchReset && state != Watcher.Event.KeeperState.SyncConnected;
			synchronized (dataWatches) {
				for (Set<Watcher> ws : dataWatches.values()) {
					result.addAll(ws);
				}
				if (clear) {
					dataWatches.clear();
				}
			}
			synchronized (existWatches) {
				for (Set<Watcher> ws : existWatches.values()) {
					result.addAll(ws);
				}
				if (clear) {
					existWatches.clear();
				}
			}
			synchronized (childWatches) {
				for (Set<Watcher> ws : childWatches.values()) {
					result.addAll(ws);
				}
				if (clear) {
					childWatches.clear();
				}
			}
			return result;
		case NodeDataChanged:
		case NodeCreated:
			synchronized (dataWatches) {
				addTo(dataWatches.remove(clientPath), result);
			}
			synchronized (existWatches) {
				addTo(existWatches.remove(clientPath), result);
			}
			break;
		case NodeChildrenChanged:
			synchronized (childWatches) {
				addTo(childWatches.remove(clientPath), result);
			}
			break;
		case NodeDeleted:
			synchronized (dataWatches) {
				addTo(dataWatches.remove(clientPath), result);
			}
			// XXX This shouldn't be needed, but just in case
			synchronized (existWatches) {
				Set<Watcher> list = existWatches.remove(clientPath);
				if (list != null) {
					addTo(existWatches.remove(clientPath), result);
					LOG.warn("We are triggering an exists watch fordelete! Shouldn't happen!");
				}
			}
			synchronized (childWatches) {
				addTo(childWatches.remove(clientPath), result);
			}
			break;
		default:
			String msg = "Unhandled watch event type " + type + " with state " + state + " on path " + clientPath;
			LOG.error(msg);
			throw new RuntimeException(msg);
		}
		return result;
	}

 

waitingEvents.add

最后一步，接近真相了waitingEvents 是 EventThread 这个线程中的阻塞队列，很明显，又是在我们第一步操作的时候实例化的一个线程。从名字可以指导，waitingEvents 是一个待处理 Watcher 的队列，EventThread 的run() 方法会不断从队列中取数据，交由 processEvent 方法处理：

public void run() {
		try {
			isRunning = true;
			while (true) { // 死循环
				Object event = waitingEvents.take(); // 从待处理的事件队列中取出事件
				if (event == eventOfDeath) {
					wasKilled = true;
				} else {
					processEvent(event); // 执行事件处理
				}
				if (wasKilled)
					synchronized (waitingEvents) {
						if (waitingEvents.isEmpty()) {
							isRunning = false;
							break;
						}
					}
			}
		} catch (InterruptedException e) {
			LOG.error("Event thread exiting due to interruption", e);
		}
		LOG.info("EventThread shut down for session: 0x{}", Long.toHexString(getSessionId()));
	}

 

ProcessEvent

由于这块的代码太长，我只把核心的代码贴出来，这里就是处理事件触发的核心代码。

private void processEvent(Object event) {
		 try {
			 if (event instanceof WatcherSetEventPair) { //判断事件类型
				 // each watcher will process the event
				 WatcherSetEventPair pair = (WatcherSetEventPair) event; //得到 watcherseteventPair
				 for (Watcher watcher : pair.watchers) { //拿到符合触发机制的所有 watcher 列表，循环进行调用
					 try {
						 watcher.process(pair.event); //调用客户端的回调 process
					 } catch (Throwable t) {
						 LOG.error("Error while calling watcher ", t);
					 }
				 }
			 }
		 }
	}