在进入今天的正题之前，先来简单介绍下Zookeeper：
  Zookeeper是一个分布式应用程序协调服务，保证数据的一致性，其提供的功能包括：配置维护、域名维护、分布式同步、组服务等。
  watch监控机制是zookeeper的关键技术之一，本文将通过zk的部分源码来简单了解下watch机制的实现原理。

watch监控机制的实现原理

  当今时代，发布订阅场景到处可见，像微信中的公众号消息订阅，或者网购场景下库存消息的订阅通知等等，这些都是属于发布订阅的场景。
  watch监控机制是zk的一个关键技术，zk通过它来实现发布订阅的功能，通过watch我们可以联想到设计模式中的观察者模式，二者确实有点类似，你可以将其看成是分布式场景下的观察者模式。

客户端watch的注册和回调

客户端watch注册实现过程：
  发送一个带有watch事件的请求——>DataWatchRegistration保存watch事件——>将请求封装成Packet并放入一个队列等待发送——>调用SendThread中的readResponse——>ZKWatchManager将该watch事件进行存储

//Zookeeper.java
    public byte[] getData(String path, Watcher watcher, Stat stat) throws KeeperException, InterruptedException {
        PathUtils.validatePath(path);
        ZooKeeper.WatchRegistration wcb = null;
        if (watcher != null) {
        	//注册watch
            wcb = new ZooKeeper.DataWatchRegistration(watcher, path);
        }
        String serverPath = this.prependChroot(path);
        RequestHeader h = new RequestHeader();
        h.setType(4);
        GetDataRequest request = new GetDataRequest();
        request.setPath(serverPath);
        request.setWatch(watcher != null);
        GetDataResponse response = new GetDataResponse();
        ReplyHeader r = this.cnxn.submitRequest(h, request, response, wcb);
        if (r.getErr() != 0) {
            throw KeeperException.create(Code.get(r.getErr()), path);
        } else {
            if (stat != null) {
                DataTree.copyStat(response.getStat(), stat);
            }
            return response.getData();
        }
    }

public ReplyHeader submitRequest(RequestHeader h, Record request, Record response, WatchRegistration watchRegistration, WatchDeregistration watchDeregistration) throws InterruptedException {
        ReplyHeader r = new ReplyHeader();
        ClientCnxn.Packet packet = this.queuePacket(h, r, request, response, (AsyncCallback)null, (String)null, (String)null, (Object)null, watchRegistration, watchDeregistration);
        synchronized(packet) {
            while(!packet.finished) {
                packet.wait();
            }
            return r;
        }
    }

    public ClientCnxn.Packet queuePacket(RequestHeader h, ReplyHeader r, Record request, Record response, AsyncCallback cb, String clientPath, String serverPath, Object ctx, WatchRegistration watchRegistration, WatchDeregistration watchDeregistration) {
        ClientCnxn.Packet packet = null;
        packet = new ClientCnxn.Packet(h, r, request, response, watchRegistration);
        packet.cb = cb;
        packet.ctx = ctx;
        packet.clientPath = clientPath;
        packet.serverPath = serverPath;
        packet.watchDeregistration = watchDeregistration;
        synchronized(this.state) {
            if (this.state.isAlive() && !this.closing) {
                if (h.getType() == -11) {
                    this.closing = true;
                }
                this.outgoingQueue.add(packet);
            } else {
                this.conLossPacket(packet);
            }
        }
        this.sendThread.getClientCnxnSocket().packetAdded();
        return packet;
    }

    class SendThread extends ZooKeeperThread {
        .....
        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) {
                if (ClientCnxn.LOG.isDebugEnabled()) {
                    ClientCnxn.LOG.debug("Got ping response for sessionid: 0x" + Long.toHexString(ClientCnxn.this.sessionId) + " after " + (System.nanoTime() - this.lastPingSentNs) / 1000000L + "ms");
                }
            } else if (replyHdr.getXid() == -4) {
                if (replyHdr.getErr() == Code.AUTHFAILED.intValue()) {
                    ClientCnxn.this.state = States.AUTH_FAILED;
                    ClientCnxn.this.eventThread.queueEvent(new WatchedEvent(EventType.None, KeeperState.AuthFailed, (String)null));
                }
                if (ClientCnxn.LOG.isDebugEnabled()) {
                    ClientCnxn.LOG.debug("Got auth sessionid:0x" + Long.toHexString(ClientCnxn.this.sessionId));
                }
            } else if (replyHdr.getXid() == -1) {
                if (ClientCnxn.LOG.isDebugEnabled()) {
                    ClientCnxn.LOG.debug("Got notification sessionid:0x" + Long.toHexString(ClientCnxn.this.sessionId));
                }
                WatcherEvent event = new WatcherEvent();
                event.deserialize(bbia, "response");
                if (ClientCnxn.this.chrootPath != null) {
                    String serverPath = event.getPath();
                    if (serverPath.compareTo(ClientCnxn.this.chrootPath) == 0) {
                        event.setPath("/");
                    } else if (serverPath.length() > ClientCnxn.this.chrootPath.length()) {
                        event.setPath(serverPath.substring(ClientCnxn.this.chrootPath.length()));
                    } else {
                        ClientCnxn.LOG.warn("Got server path " + event.getPath() + " which is too short for chroot path " + ClientCnxn.this.chrootPath);
                    }
                }
                WatchedEvent we = new WatchedEvent(event);
                if (ClientCnxn.LOG.isDebugEnabled()) {
                    ClientCnxn.LOG.debug("Got " + we + " for sessionid 0x" + Long.toHexString(ClientCnxn.this.sessionId));
                }
                ClientCnxn.this.eventThread.queueEvent(we);
            } else if (this.tunnelAuthInProgress()) {
                GetSASLRequest request = new GetSASLRequest();
                request.deserialize(bbia, "token");
                ClientCnxn.this.zooKeeperSaslClient.respondToServer(request.getToken(), ClientCnxn.this);
            } else {
                ClientCnxn.Packet packet;
                synchronized(ClientCnxn.this.pendingQueue) {
                    if (ClientCnxn.this.pendingQueue.size() == 0) {
                        throw new IOException("Nothing in the queue, but got " + replyHdr.getXid());
                    }
                    packet = (ClientCnxn.Packet)ClientCnxn.this.pendingQueue.remove();
                }
                try {
                    if (packet.requestHeader.getXid() != replyHdr.getXid()) {
                        packet.replyHeader.setErr(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() > 0L) {
                        ClientCnxn.this.lastZxid = replyHdr.getZxid();
                    }
                    if (packet.response != null && replyHdr.getErr() == 0) {
                        packet.response.deserialize(bbia, "response");
                    }
                    if (ClientCnxn.LOG.isDebugEnabled()) {
                        ClientCnxn.LOG.debug("Reading reply sessionid:0x" + Long.toHexString(ClientCnxn.this.sessionId) + ", packet:: " + packet);
                    }
                } finally {
                    ClientCnxn.this.finishPacket(packet);
                }
            }
        }

private void finishPacket(ClientCnxn.Packet p) {
        int err = p.replyHeader.getErr();
        if (p.watchRegistration != null) {
            p.watchRegistration.register(err);
        }
        if (p.watchDeregistration != null) {
            Map materializedWatchers = null;
            try {
                materializedWatchers = p.watchDeregistration.unregister(err);
                Iterator i$ = materializedWatchers.entrySet().iterator();
                while(i$.hasNext()) {
                    Entry<EventType, Set<Watcher>> entry = (Entry)i$.next();
                    Set<Watcher> watchers = (Set)entry.getValue();
                    if (watchers.size() > 0) {
                        this.queueEvent(p.watchDeregistration.getClientPath(), err, watchers, (EventType)entry.getKey());
                        p.replyHeader.setErr(Code.OK.intValue());
                    }
                }
            } catch (NoWatcherException var9) {
                LOG.error("Failed to find watcher!", var9);
                p.replyHeader.setErr(var9.code().intValue());
            } catch (KeeperException var10) {
                LOG.error("Exception when removing watcher", var10);
                p.replyHeader.setErr(var10.code().intValue());
            }
        }
        if (p.cb == null) {
            synchronized(p) {
                p.finished = true;
                p.notifyAll();
            }
        } else {
            p.finished = true;
            this.eventThread.queuePacket(p);
        }
    }

客户端回调处理过程：
  在SendThread.readResponse()中的xid=-1来进行处理——>调用 eventThread.queueEvent()进行处理

    class SendThread extends ZooKeeperThread {
        .....
        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) {
  				....
            } else if (replyHdr.getXid() == -1) {
                if (ClientCnxn.LOG.isDebugEnabled()) {
                    ClientCnxn.LOG.debug("Got notification sessionid:0x" + Long.toHexString(ClientCnxn.this.sessionId));
                }
                WatcherEvent event = new WatcherEvent();
                event.deserialize(bbia, "response");
                if (ClientCnxn.this.chrootPath != null) {
                    String serverPath = event.getPath();
                    if (serverPath.compareTo(ClientCnxn.this.chrootPath) == 0) {
                        event.setPath("/");
                    } else if (serverPath.length() > ClientCnxn.this.chrootPath.length()) {
                        event.setPath(serverPath.substring(ClientCnxn.this.chrootPath.length()));
                    } else {
                        ClientCnxn.LOG.warn("Got server path " + event.getPath() + " which is too short for chroot path " + ClientCnxn.this.chrootPath);
                    }
                }
                WatchedEvent we = new WatchedEvent(event);
                if (ClientCnxn.LOG.isDebugEnabled()) {
                    ClientCnxn.LOG.debug("Got " + we + " for sessionid 0x" + Long.toHexString(ClientCnxn.this.sessionId));
                }
                ClientCnxn.this.eventThread.queueEvent(we);
            }
            ....
        }

服务端watch的注册和触发

服务端watch注册实现过程：
  判断请求是否带有watch注册事件——>通过FinalRequestProcessor中的processRequest解析请求——>若有watch，则调用zks.getZKDatabase().getData将事件注册到WatchManager

//FinalRequestProcessor.java
    public void processRequest(Request request) {
		....
        ProcessTxnResult rc = null;
        synchronized(this.zks.outstandingChanges) {
            rc = this.zks.processTxn(request);
            if (request.getHdr() != null) {
                TxnHeader hdr = request.getHdr();
                Record txn = request.getTxn();
                long zxid = hdr.getZxid();
                while(!this.zks.outstandingChanges.isEmpty() && ((ChangeRecord)this.zks.outstandingChanges.get(0)).zxid <= zxid) {
                    ChangeRecord cr = (ChangeRecord)this.zks.outstandingChanges.remove(0);
                    if (cr.zxid < zxid) {
                        LOG.warn("Zxid outstanding " + cr.zxid + " is less than current " + zxid);
                    }
                    if (this.zks.outstandingChangesForPath.get(cr.path) == cr) {
                        this.zks.outstandingChangesForPath.remove(cr.path);
                    }
                }
            }
            if (request.isQuorum()) {
                this.zks.getZKDatabase().addCommittedProposal(request);
            }
        }
        if (request.type != -11 || !this.connClosedByClient(request) || !this.closeSession(this.zks.serverCnxnFactory, request.sessionId) && !this.closeSession(this.zks.secureServerCnxnFactory, request.sessionId)) {
            if (request.cnxn != null) {
                ServerCnxn cnxn = request.cnxn;
                String lastOp = "NA";
                this.zks.decInProcess();
                Code err = Code.OK;
                Object rsp = null;
                try {
                    if (request.getHdr() != null && request.getHdr().getType() == -1) {
                        if (request.getException() != null) {
                            throw request.getException();
                        }
                        throw KeeperException.create(Code.get(((ErrorTxn)request.getTxn()).getErr()));
                    }
                    KeeperException ke = request.getException();
                    if (ke != null && request.type != 14) {
                        throw ke;
                    }
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("{}", request);
                    }
                    boolean removed;
                    String msg;
                    WatcherType type;
                    Stat stat;
                    DataNode n;
                    List children;
                    Stat stat;
                    label170:
                    switch(request.type) {
                    ...
                    case 4:
                        lastOp = "GETD";
                        GetDataRequest getDataRequest = new GetDataRequest();
                        ByteBufferInputStream.byteBuffer2Record(request.request, getDataRequest);
                        n = this.zks.getZKDatabase().getNode(getDataRequest.getPath());
                        if (n == null) {
                            throw new NoNodeException();
                        }
                        PrepRequestProcessor.checkACL(this.zks, this.zks.getZKDatabase().aclForNode(n), 1, request.authInfo);
                        stat = new Stat();
                        byte[] b = this.zks.getZKDatabase().getData(getDataRequest.getPath(), stat, getDataRequest.getWatch() ? cnxn : null);
                        rsp = new GetDataResponse(b, stat);
                        break;
                    ....
                } catch (SessionMovedException var15) {
                    ....
                }
                long lastZxid = this.zks.getZKDatabase().getDataTreeLastProcessedZxid();
                ReplyHeader hdr = new ReplyHeader(request.cxid, lastZxid, err.intValue());
                this.zks.serverStats().updateLatency(request.createTime);
                cnxn.updateStatsForResponse((long)request.cxid, lastZxid, lastOp, request.createTime, Time.currentElapsedTime());
                try {
                    cnxn.sendResponse(hdr, (Record)rsp, "response");
                    if (request.type == -11) {
                        cnxn.sendCloseSession();
                    }
                } catch (IOException var14) {
                    LOG.error("FIXMSG", var14);
                }
            }
        }
    }

服务端watch事件的触发过程：
  setData（）对节点数据变更——>调用 WatchManager.triggerWatch 触发事件——>触发之后删除(客户端watch机制是一次性的)

//DataTree.java
    public Stat setData(String path, byte[] data, int version, long zxid, long time) throws NoNodeException {
        Stat s = new Stat();
        DataNode n = (DataNode)this.nodes.get(path);
        if (n == null) {
            throw new NoNodeException();
        } else {
            byte[] lastdata = null;
            byte[] lastdata;
            synchronized(n) {
                lastdata = n.data;
                n.data = data;
                n.stat.setMtime(time);
                n.stat.setMzxid(zxid);
                n.stat.setVersion(version);
                n.copyStat(s);
            }
            String lastPrefix = this.getMaxPrefixWithQuota(path);
            if (lastPrefix != null) {
                this.updateBytes(lastPrefix, (long)((data == null ? 0 : data.length) - (lastdata == null ? 0 : lastdata.length)));
            }
            this.dataWatches.triggerWatch(path, EventType.NodeDataChanged);
            return s;
        }
    }

//WatchManager.java
    Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {
        WatchedEvent e = new WatchedEvent(type, KeeperState.SyncConnected, path);
        HashSet watchers;
        synchronized(this) {
            watchers = (HashSet)this.watchTable.remove(path);
            if (watchers == null || watchers.isEmpty()) {
                if (LOG.isTraceEnabled()) {
                    ZooTrace.logTraceMessage(LOG, 64L, "No watchers for " + path);
                }
                return null;
            }
            Iterator i$ = watchers.iterator();
            while(i$.hasNext()) {
                Watcher w = (Watcher)i$.next();
                HashSet<String> paths = (HashSet)this.watch2Paths.get(w);
                if (paths != null) {
                    paths.remove(path);
                }
            }
        }
        Iterator i$ = watchers.iterator();
        while(true) {
            Watcher w;
            do {
                if (!i$.hasNext()) {
                    return watchers;
                }
                w = (Watcher)i$.next();
            } while(supress != null && supress.contains(w));
            w.process(e);
        }
    }

//DataTree.java
    public void deleteNode(String path, long zxid) throws NoNodeException {
        int lastSlash = path.lastIndexOf(47);
        String parentName = path.substring(0, lastSlash);
        String childName = path.substring(lastSlash + 1);
        DataNode node = (DataNode)this.nodes.get(path);
        if (node == null) {
            throw new NoNodeException();
        } else {
            this.nodes.remove(path);
            synchronized(node) {
                this.aclCache.removeUsage(node.acl);
            }
            DataNode parent = (DataNode)this.nodes.get(parentName);
            if (parent == null) {
                throw new NoNodeException();
            } else {
                synchronized(parent) {
                    parent.removeChild(childName);
                    parent.stat.setPzxid(zxid);
                    long eowner = node.stat.getEphemeralOwner();
                    EphemeralType ephemeralType = EphemeralType.get(eowner);
                    if (ephemeralType == EphemeralType.CONTAINER) {
                        this.containers.remove(path);
                    } else if (ephemeralType == EphemeralType.TTL) {
                        this.ttls.remove(path);
                    } else if (eowner != 0L) {
                        HashSet<String> nodes = (HashSet)this.ephemerals.get(eowner);
                        if (nodes != null) {
                            synchronized(nodes) {
                                nodes.remove(path);
                            }
                        }
                    }
                }
                if (parentName.startsWith("/zookeeper") && "zookeeper_limits".equals(childName)) {
                    this.pTrie.deletePath(parentName.substring("/zookeeper/quota".length()));
                }
                String lastPrefix = this.getMaxPrefixWithQuota(path);
                if (lastPrefix != null) {
                    this.updateCount(lastPrefix, -1);
                    int bytes = false;
                    int bytes;
                    synchronized(node) {
                        bytes = node.data == null ? 0 : -node.data.length;
                    }
                    this.updateBytes(lastPrefix, (long)bytes);
                }
                if (LOG.isTraceEnabled()) {
                    ZooTrace.logTraceMessage(LOG, 64L, "dataWatches.triggerWatch " + path);
                    ZooTrace.logTraceMessage(LOG, 64L, "childWatches.triggerWatch " + parentName);
                }
                Set<Watcher> processed = this.dataWatches.triggerWatch(path, EventType.NodeDeleted);
                this.childWatches.triggerWatch(path, EventType.NodeDeleted, processed);
                this.childWatches.triggerWatch("".equals(parentName) ? "/" : parentName, EventType.NodeChildrenChanged);
            }
        }
    }

总结

  客户端和服务端各自处理watch事件，并将所需信息分别在两端，这样一来，能够减少彼此之间的通信内容和频率，大大提升了服务的处理性能。