Watcher--数据变更的通知

青旅半醒 2022-12-17 05:18 154阅读 0赞

Zookeeper提供了分布式数据的发布/订阅功能，Zookeeper引入了Watcher机制来实现分布式的通知功能，Zookeeper允许客户端向服务端注册一个Watcher监听，当服务端的一些指定事件触发了这个Watcher，则会通知客户端进行相应的处理，Watcher机制概述如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ljYjE5MTQ4NDU0NTE_size_16_color_FFFFFF_t_70][]

从上图可以看出，客户端在向Zookeeper注册Watcher时会同步把该Watcher存储在WatcherManger中，当Zookeeper服务器端触发Watcher时，会通知客户端，然后从WatcherManger中取出对应的Watcher执行相应的回调逻辑。

**Watcher接口**

Watcher接口用于表示一个标准的事件处理器，里面就一个抽象方法

abstract public void process(WatchedEvent event);

**WatchedEvent**

/**
     *  A WatchedEvent represents a change on the ZooKeeper that a Watcher
     *  is able to respond to.  The WatchedEvent includes exactly what happened,
     *  the current state of the ZooKeeper, and the path of the znode that
     *  was involved in the event.
     */
    public class WatchedEvent {
        //事件触发时Zookeeper的状态
        final private KeeperState keeperState;
       
        //触发的事件类型
        final private EventType eventType;
    
        //事件涉及到的节点路径
        private String path;
    
        //省略一些方法
    
        /**
         *  Convert WatchedEvent to type that can be sent over network
         */
        public WatcherEvent getWrapper() {
            return new WatcherEvent(eventType.getIntValue(), 
                                    keeperState.getIntValue(), 
                                    path);
        }
    }

**WatchedEvent和WatcherEvent的区别和联系**

本质上讲两者是同一个事物，都是对一个服务端事件的封装，区别是WatchedEvent是一个逻辑事件，用于客户端与服务端执行过程中的逻辑对象，而WatcherEvent实现了序列化接口，可以用于网络传输，WatchedEvent类提供了getWapper方法把WatchedEvent转换为WatcherEvent。

不管是WatcherEvent，还是WatchedEvent，其对Zookeeper服务端事件的封装是及其简单的，只会包含事件触发时服务端的状态，具体事件类型，涉及到的节点的路径。

如：**KeeperState**：SyncConnected

**EventType**：NodeChanged

**Path**：/test

至于节点的数据由什么变为什么，需要重新加载一次，从Watcher通知里面是无法得知的。

**KeeperState，事件发生时Zookeeper的状态**

/**
             * Enumeration of states the ZooKeeper may be at the event
             */
            public enum KeeperState {
                /** Unused, this state is never generated by the server */
                @Deprecated
                Unknown (-1),
    
                /** The client is in the disconnected state - it is not connected
                 * to any server in the ensemble. */
                Disconnected (0),
    
                /** Unused, this state is never generated by the server */
                @Deprecated
                NoSyncConnected (1),
    
                /** The client is in the connected state - it is connected
                 * to a server in the ensemble (one of the servers specified
                 * in the host connection parameter during ZooKeeper client
                 * creation). */
                SyncConnected (3),
    
                /**
                 * Auth failed state
                 */
                AuthFailed (4),
    
                /**
                 * The client is connected to a read-only server, that is the
                 * server which is not currently connected to the majority.
                 * The only operations allowed after receiving this state is
                 * read operations.
                 * This state is generated for read-only clients only since
                 * read/write clients aren't allowed to connect to r/o servers.
                 */
                ConnectedReadOnly (5),
    
                /**
                  * SaslAuthenticated: used to notify clients that they are SASL-authenticated,
                  * so that they can perform Zookeeper actions with their SASL-authorized permissions.
                  */
                SaslAuthenticated(6),
    
                /** The serving cluster has expired this session. The ZooKeeper
                 * client connection (the session) is no longer valid. You must
                 * create a new client connection (instantiate a new ZooKeeper
                 * instance) if you with to access the ensemble. */
                Expired (-112);
    }

EventType：Zookeeper上事件的类型

/**
             * Enumeration of types of events that may occur on the ZooKeeper
             */
            public enum EventType {
                None (-1),
                NodeCreated (1),
                NodeDeleted (2),
                NodeDataChanged (3),
                NodeChildrenChanged (4);
    }

注意NodeDataChanged表示dataVersion的表更。NodeChildrenChanged表示的节点的子节点列表发生变更，即新增或者删除子节点，但是子节点的数据发生变化是不会触发的。

**客户端是如何注册Watcher的？**

创建一个Zookeeper时可以向构造函数中传入一个默认的Watcher，构造函数如下（省略其中一些构造函数）

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
                boolean canBeReadOnly)
            throws IOException
        {
            //设置默认的Watcher
            watchManager.defaultWatcher = watcher;
            
            //地址解析器解析地址
            ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
    
            //记录Zookeeper地址相关的信息
            HostProvider hostProvider = new StaticHostProvider(
                    connectStringParser.getServerAddresses());
    
            //管理着客户端的socket i/o
            cnxn = new ClientCnxn(connectStringParser.getChrootPath(),
                    hostProvider, sessionTimeout, this, watchManager,
                    getClientCnxnSocket(), canBeReadOnly);
            cnxn.start();
        }

**设置默认的watcher到ZKWatcherManger的defaultWatcher中**，除了通过这种方式注册watcher之外，还可以通过 **getData、exists和 getChildren**三个接口进行注册Watcher。采用哪种方式注册逻辑都是一样的，此处使用getData进行说明。

getData接口的定义如下（去除了抛出的异常等信息）

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

在调用getData接口注册Watcher时详细逻辑如下：

public byte[] getData(final String path, Watcher watcher, Stat stat)
            throws KeeperException, InterruptedException
         {
            final String clientPath = path;
    
            //验证地址的合法性
            PathUtils.validatePath(clientPath);
    
            // the watch contains the un-chroot path
            WatchRegistration wcb = null;
            if (watcher != null) {
                wcb = new DataWatchRegistration(watcher, clientPath);
            }
    
            //解析获得服务端地址，主要是前缀处理
            final String serverPath = prependChroot(clientPath);
    
            //请求头
            RequestHeader h = new RequestHeader();
    
            //设置类型
            h.setType(ZooDefs.OpCode.getData);
    
            //设置请求使用的协议，后续会进行介绍
            GetDataRequest request = new GetDataRequest();
    
            //设置服务端路径
            request.setPath(serverPath);
    
            //设置是否注册watcher，只是打一个标识，并没有真正的传递Watcher
            request.setWatch(watcher != null);
            
            //响应体
            GetDataResponse response = new GetDataResponse();
            
            //获得响应头，通过网络调用获得结果并进行后续的处理
            ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
            if (r.getErr() != 0) {
                throw KeeperException.create(KeeperException.Code.get(r.getErr()),
                        clientPath);
            }
            if (stat != null) {
                DataTree.copyStat(response.getStat(), stat);
            }
            return response.getData();
        }

从上面的代码中可以看出，最终的调用就是

//Zookeeper中的调用处
    ReplyHeader r = cnxn.submitRequest(h,request,response,wcb);
    
    
    //ClientCnxn中的方法submitRequest
    public ReplyHeader submitRequest(RequestHeader h, Record request,
                Record response, WatchRegistration watchRegistration)
                throws InterruptedException {
            ReplyHeader r = new ReplyHeader();
            Packet packet = queuePacket(h, r, request, response, null, null, null,
                        null, watchRegistration);
            synchronized (packet) {
                while (!packet.finished) {
                    packet.wait();
                }
            }
            return r;
        }
    
    
    //Packet中的queuePacket方法
    Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
                Record response, AsyncCallback cb, String clientPath,
                String serverPath, Object ctx, WatchRegistration watchRegistration)
        {
            Packet packet = null;
    
            //此处还没有为Packet生成Xid，在发送的时候生成，在接口ClientCnxnSocket::doIO()
            synchronized (outgoingQueue) {
                packet = new Packet(h, r, request, response, watchRegistration);
                packet.cb = cb;
                packet.ctx = ctx;
                packet.clientPath = clientPath;
                packet.serverPath = serverPath;
                if (!state.isAlive() || closing) {
                    conLossPacket(packet);
                } else {
                    // If the client is asking to close the session then
                    // mark as closing
                    if (h.getType() == OpCode.closeSession) {
                        closing = true;
                    }
                    
                    //加入发送队列中等待发送
                    outgoingQueue.add(packet);
                }
            }
            sendThread.getClientCnxnSocket().wakeupCnxn();
            return packet;
        }

在Zookeeper中，Packet可以被看做一个最小的通信协议单元，用于进行客户端与服务端之间的通信传输，任何需要传输的对象都需要被包装为一个Packet对象，同时WatchRegistration 也被包装到Packet对象中，然后放入发送队列中等待发送。

随后Zookeeper的客户端发送这个请求，并等待请求的响应，完成请求后返回，会由SendThread的readResponse进行响应。finishPacket方法会从Packet中取出对应的Watcher并注册到ZKWatcherManager中。

private void finishPacket(Packet p) {
            if (p.watchRegistration != null) {
    
                //注册到ZKWatcherManager中
                p.watchRegistration.register(p.replyHeader.getErr());
            }
    
            if (p.cb == null) {
                synchronized (p) {
                    p.finished = true;
                    p.notifyAll();
                }
            } else {
                p.finished = true;
                
                //事件线程进行处理
                eventThread.queuePacket(p);
            }
        }
    
       //WatchRegistration.register方法如下
       /**
             * Register the watcher with the set of watches on path.
             * @param rc the result code of the operation that attempted to
             * add the watch on the path.
             */
            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);
                        }
                        watchers.add(watcher);
                    }
                }
            }

注册流程图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ljYjE5MTQ4NDU0NTE_size_16_color_FFFFFF_t_70 1][]

注：客户端注册的Watcher是不会发送到服务端的。可以从Packet的createBB方法中看到序列化时没有处理Watcher，代码如下：

public void createBB() {
                try {
                    ByteArrayOutputStream baos = new ByteArrayOutputStream();
                    BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
                    boa.writeInt(-1, "len"); // We'll fill this in later
                    if (requestHeader != null) {
                        //序列化header
                        requestHeader.serialize(boa, "header");
                    }
                    if (request instanceof ConnectRequest) {
                        request.serialize(boa, "connect");
                        boa.writeBool(readOnly, "readOnly");
                    } else if (request != null) {
    
                        //序列化request
                        request.serialize(boa, "request");
                    }
                    baos.close();
                    this.bb = ByteBuffer.wrap(baos.toByteArray());
                    this.bb.putInt(this.bb.capacity() - 4);
                    this.bb.rewind();
                } catch (IOException e) {
                    LOG.warn("Ignoring unexpected exception", e);
                }
            }

可以看到只会处理header和request（request中有是否注册Watcher的标志），对于Watcher，没有序列化，当然也不会传输到服务端。

**但是Watcher不传到服务端，服务端是怎么知道有Watcher存在的？又是怎么实现通知的？**

Zookeeper服务端处理Watcher的序列图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ljYjE5MTQ4NDU0NTE_size_16_color_FFFFFF_t_70 2][]

FinalRequestProcessor.processRequest()中会判断当前请求是否需要注册Watcher，这个是由客户端传递过来时确定的

case OpCode.getData: {
                    /*
                     *此处省略掉部分代码
                     */
                     //getDataRequest.getWatch() 为true表示需要注册Watcher
                    byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,
                            getDataRequest.getWatch() ? cnxn : null);
                    rsp = new GetDataResponse(b, stat);
                    break;
                }

如果需要注册Watcher，于是就会将当前的ServerCnxn进行注册，我们看下ServerCnxn的定义

/**
     * Interface to a Server connection - represents a connection from a client
     * to the server.
     * 一个Server连接，代表一个客户端与服务端的连接
     */
    public abstract class ServerCnxn implements Stats, Watcher {
        //省略部分代码
    
        public abstract void process(WatchedEvent event);
    
        //省略部分代码
    }

服务端的Watcher统一由WatchManager进行管理

/**
     * This class manages watches. It allows watches to be associated with a string
     * and removes watchers and their watches in addition to managing triggers.
     */
    public class WatchManager {
        private static final Logger LOG = LoggerFactory.getLogger(WatchManager.class);
    
        //从数据节点的角度来托管Watcher，key表示path，Value表示该节点注册的Watcher
        private final HashMap<String, HashSet<Watcher>> watchTable =
            new HashMap<String, HashSet<Watcher>>();
    
        //从Watcher的粒度来控制事件触发时需要触发的数据节点
        private final HashMap<Watcher, HashSet<String>> watch2Paths =
            new HashMap<Watcher, HashSet<String>>();
    
        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);
        }
        //省略部分代码
    }

**Watcher的触发**

上面了解了Watcher的注册，会将ServerCnxn进行注册到WatcherManager中，前面了解过事件的类型（EventType）可以是NodeDataChanged，NodeDataChanged的触发条件是“Watcher监听的数据节点的内容发生变更”。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;
            if((lastPrefix = getMaxPrefixWithQuota(path)) != null) {
              this.updateBytes(lastPrefix, (data == null ? 0 : data.length)
                  - (lastdata == null ? 0 : lastdata.length));
            }
            
            //触发Watcher
            dataWatches.triggerWatch(path, EventType.NodeDataChanged);
            return s;
        }

设置数据后调用WatchManager的triggerWatch方法来触发Watcher

public Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {
            //1、事件封装
            WatchedEvent e = new WatchedEvent(type,KeeperState.SyncConnected, path);
            HashSet<Watcher> watchers;
            synchronized (this) {
                //2、查询Watcher并移除Watcher，所以可以看出Watcher是一次性的，使用后会失效
                watchers = watchTable.remove(path);
                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) {
                    HashSet<String> paths = watch2Paths.get(w);
                    if (paths != null) {
                        paths.remove(path);
                    }
                }
            }
            for (Watcher w : watchers) {
                if (supress != null && supress.contains(w)) {
                    continue;
                }
                
                //3、调用Watcher接口的process方法处理
                w.process(e);
            }
            return watchers;
        }

可以看出Watcher是一次性的，使用后会失效。

前面说过注册时其实注册的就是ServerCnxn，ServerCnxn的实现有NIOServerCnxn 和 NettyServerCnxn....。实现逻辑都差不多，此处以NIOServerCnxn进行说明。

//此方法经过处理，去除掉日志等
    public void process(WatchedEvent event) {
            //参数依次是 xid zxid err ，将ReplyHeader中的参数设置为-1，标识这是一个通知 
            ReplyHeader h = new ReplyHeader(-1, -1L, 0);
    
            //WatchedEvent包装为WatcherEvent，方便进行序列化传输
            WatcherEvent e = event.getWrapper();
    
            try {
                //发送响应回客户端
                sendResponse(h, e, "notification");
            } catch (IOException e1) {
                close();
            }
        }

**客户端回掉Watcher**

由SendThread的readResponse方法进行接收。

void readResponse(ByteBuffer incomingBuffer) throws IOException {
                ByteBufferInputStream bbis = new ByteBufferInputStream(
                        incomingBuffer);
                BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
                ReplyHeader replyHdr = new ReplyHeader();
                
                //反序列化获得头部
                replyHdr.deserialize(bbia, "header");
    
                //***省略了部分代码***//            
                // -1 标识这是一个通知
                if (replyHdr.getXid() == -1) { 
                   
                    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.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
                    WatchedEvent we = new WatchedEvent(event);
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("Got " + we + " for sessionid 0x"
                                + Long.toHexString(sessionId));
                    }
                    
                    //加入Eventqueue进行后续的回掉处理
                    eventThread.queueEvent( we );
                    return;
                }
                //*******省略部分代码*******//
            }

**EventThread的queueEvent方法如下**

public void queueEvent(WatchedEvent event) {
                if (event.getType() == EventType.None
                        && sessionState == event.getState()) {
                    return;
                }
                
                //获得session的状态。
                sessionState = event.getState();
    
                // materialize the watchers based on the event
                WatcherSetEventPair pair = new WatcherSetEventPair(
                        watcher.materialize(event.getState(), event.getType(),
                                event.getPath()),
                                event);
                // queue the pair (watch set & event) for later processing
                waitingEvents.add(pair);
            }

可以看出，最核心的就是ZKWatcherManager的**materialize**方法

接口定义如下：

/**
     */
    public interface ClientWatchManager {
        /**
         * Return a set of watchers that should be notified of the event. The 
         * manager must not notify the watcher(s), however it will update it's 
         * internal structure as if the watches had triggered. The intent being 
         * that the callee is now responsible for notifying the watchers of the 
         * event, possibly at some later time.
         * 
         * @param state event state
         * @param type event type
         * @param path event path
         * @return may be empty set but must not be null
         */
        public Set<Watcher> materialize(Watcher.Event.KeeperState state,
            Watcher.Event.EventType type, String path);
    }

ZKWatcherManager中materialize实现如下：

@Override
            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;
            }
        }

客户端在识别出事件类型EventType后，会从相应的Watcher存储（即dataWatchers、existsWatchers或childWatchers中的一个或多个）中去除相应的Watcher。使用remove接口，表示客服端的Watcher也是一次性的。解析完Watcher之后就会放入到EventThread的waitingEvents中，等待线程消耗处理。

EventThread的run方法如下：**去除了日志等**

@Override
            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) {
    
               }
            }

processEvent方法如下

private void processEvent(Object event) {
              try {
                  if (event instanceof WatcherSetEventPair) {
                      WatcherSetEventPair pair = (WatcherSetEventPair) event;
                      for (Watcher watcher : pair.watchers) {
                          try {
                              //调用process进行处理
                              watcher.process(pair.event);
                          } catch (Throwable t) {
                              
                          }
                      }
                  } 
                  else{
                            //去除
                      }
        }

从代码可知，EventThread线程每次从waitingEvents队列中取出一个Watcher，并进行串行同步处理。此Watcher才是客户端真正注册的Watcher。

**Watcher特性总结**

一次性：不管是客户端的Watcher还是服务端的Watcher，都只能使用一次，使用一次后就失效

客户端串行化：客户端Watcher回调的过程是一个串行同步的过程

轻量：WatchedEvent是Zookeeper整个Watcher通知机制的最小通知单元，这个数据只包含三个部分：通知状态，事件类型和节点路劲。并不会有事件的详细类型。这样设计客户端注册时也不需要把整个Watcher传递到服务端。

此文仅供个人学习记录使用。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ljYjE5MTQ4NDU0NTE_size_16_color_FFFFFF_t_70]: /images/20221123/f5c9622ccb6c40138cc4fbc27fb8ad73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ljYjE5MTQ4NDU0NTE_size_16_color_FFFFFF_t_70 1]: /images/20221123/20985837c83c42519983a360bb577c4a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ljYjE5MTQ4NDU0NTE_size_16_color_FFFFFF_t_70 2]: /images/20221123/ba903f35f7034b148b42c3485f3a5b65.png