搞懂ZooKeeper的Watcher之源码分析及特性总结

前言

　　本章讲ZooKeeper重要的机制，Watcher特性。ZooKeeper允许客户端向服务端注册Watcher监听，当服务端一些指定事件触发了这个Watcher，那么就会向指定客户端发送一个事件通知客户端执行回调逻辑

一.Watcher机制

　　ZooKeeper允许客户端向服务端注册感兴趣的Watcher监听，当服务端触发了这个Watcher，那么就会向客户端发送一个时间来实现分布式的通知功能。真正的Watcher回调与业务逻辑执行都在客户端

　　那么需要注意一下，给客户端的通知里只会告诉你通知状态（KeeperState），事件类型（EventType）和路径（Path）。不会告诉你原始数据和更新过后的数据！

　　Watcher机制包括三部分：注册、存储、通知

1. 注册：注册Watcher
2. 存储：讲Watcher对象存在客户端的WatcherManager中
3. 通知：服务端触发Watcher事件，通知客户端，客户端从WatcherManager中取出对应的Watcher对象执行回调 

　　那么接下来，我们就分这3步来分析：

注册

　　我们可以通过以下方式向服务端注册Watcher，主要是构造参数、getData、getChildren和exists方法：

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
public byte[] getData(String path, Watcher watcher, Stat stat)
public List<String> getChildren(String path, Watcher watcher)
public Stat exists(String path, Watcher watcher)

　　我们就看getData方法，从源码角度看看如何注册的，可以看到首先封装了一个WatchRegistration对象，保存了节点的路径和Watcher对象的关系，然后在请求的request设置了是否有watcher这么一个boolean的成员变量：

public byte[] getData(String path, Watcher watcher, Stat stat) throws KeeperException, InterruptedException {
    PathUtils.validatePath(path);

    // 封装一个WatcherRegistration的对象，保存节点路径和Watcher的对应关系
    ZooKeeper.WatchRegistration wcb = null;
    if (watcher != null) {
        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);

    // 标记是否有watcher
    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();
    }
}

class DataWatchRegistration extends ZooKeeper.WatchRegistration {
    // 保存节点路径和Watcher的关系
    public DataWatchRegistration(Watcher watcher, String clientPath) {
        super(watcher, clientPath);
    }

    ...
}

abstract class WatchRegistration {
    private Watcher watcher;
    private String clientPath;

    public WatchRegistration(Watcher watcher, String clientPath) {
        this.watcher = watcher;
        this.clientPath = clientPath;
    }
    ...
}

　　然后我们继续接着看这个wcb变量做了什么（已经用紫色标注该变量），简单来说就是这个变量被封装在了packet对象里，packet可以看成一个最小的通信协议单元，传输信息。最后将packet对象放到了发送队列里SendThread里

public ReplyHeader submitRequest(RequestHeader h, Record request, Record response, WatchRegistration watchRegistration) throws InterruptedException {
    ReplyHeader r = new ReplyHeader();
    // 客户端与服务端的网络传输
    ClientCnxn.Packet packet = this.queuePacket(h, r, request, response, (AsyncCallback)null, (String)null, (String)null, (Object)null, watchRegistration);
    synchronized(packet) {
        while(!packet.finished) {
            packet.wait();
        }
        return r;
    }
}

ClientCnxn.Packet queuePacket(RequestHeader h, ReplyHeader r, Record request, Record response, AsyncCallback cb, String clientPath, String serverPath, Object ctx, WatchRegistration watchRegistration) {
    ClientCnxn.Packet packet = null;
    LinkedList var11 = this.outgoingQueue;
    synchronized(this.outgoingQueue) {
        // 任何传输的对象都包装成Packet对象
        packet = new ClientCnxn.Packet(h, r, request, response, watchRegistration);
        packet.cb = cb;
        packet.ctx = ctx;
        packet.clientPath = clientPath;
        packet.serverPath = serverPath;
        if (this.state.isAlive() && !this.closing) {
            if (h.getType() == -11) {
                this.closing = true;
            }

            // 放入发送队列中，等待发送
            this.outgoingQueue.add(packet);
        } else {
            this.conLossPacket(packet);
        }
    }

    this.sendThread.getClientCnxnSocket().wakeupCnxn();
    return packet;
}

　　然后我们看org.apache.zookeeper.ClientCnxnSocketNIO#doIO这个方法，关键代码已经用红色标注出来了，从要发送的队列outgoingQueue中取出packet然后序列化到底层数组，注意了，这里没有序列化前面说的WatchRegistration对象，只序列化了requestHeader和request两个属性，也就是说，服务端并不会接收到阶段路径和watcher对象的关系，回调的业务逻辑代码也不会给服务端！

void doIO(List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn cnxn) throws InterruptedException, IOException {
    SocketChannel sock = (SocketChannel)this.sockKey.channel();
    if (sock == null) {
        throw new IOException("Socket is null!");
    } else {
        // 是否可读
        if (this.sockKey.isReadable()) {
            ...
        }

        if (this.sockKey.isWritable()) {
            synchronized(outgoingQueue) {
                Packet p = this.findSendablePacket(outgoingQueue, cnxn.sendThread.clientTunneledAuthenticationInProgress());
                if (p != null) {
                    this.updateLastSend();
                    if (p.bb == null) {
                        if (p.requestHeader != null && p.requestHeader.getType() != 11 && p.requestHeader.getType() != 100) {
                            p.requestHeader.setXid(cnxn.getXid());
                        }

                        // 序列化
                        p.createBB();
                    }
                    
                    sock.write(p.bb);
                    ...
                }

                ...
            }
        }
    }
}

public void createBB() {
    try {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
        boa.writeInt(-1, "len");

        // 序列化header
        if (this.requestHeader != null) {
            this.requestHeader.serialize(boa, "header");
        }
        if (this.request instanceof ConnectRequest) {
            this.request.serialize(boa, "connect");
            boa.writeBool(this.readOnly, "readOnly");

            // 序列化request
        } else if (this.request != null) {
            this.request.serialize(boa, "request");
        }
        baos.close();
        this.bb = ByteBuffer.wrap(baos.toByteArray());
        this.bb.putInt(this.bb.capacity() - 4);
        this.bb.rewind();
    } catch (IOException var3) {
        ClientCnxn.LOG.warn("Ignoring unexpected exception", var3);
    }
}

 

存储

　　上面都是客户端发起请求的过程，那么接下来我们看服务端接收到请求会做些什么，ZooKeeper的服务端对于客户端的请求，采用了典型的责任链模式，也就是说客户端的每个请求都由几个不同的处理器来依次进行处理，我们这里就看这个方法:org.apache.zookeeper.server.FinalRequestProcessor#processRequest

public void processRequest(Request request) {
    ...
    PrepRequestProcessor.checkACL(this.zks, this.zks.getZKDatabase().convertLong(aclG), 1, request.authInfo);
    Stat stat = new Stat();
    // 这里根据客户端设置的是否有watch变量来传入watcher对象
    // 如果true则将当前的ServerCnxn传入（ServerCnxn代表客户端和服务端的连接）
    byte[] b = this.zks.getZKDatabase().getData(getDataRequest.getPath(), stat, getDataRequest.getWatch() ? cnxn : null);
    rsp = new GetDataResponse(b, stat);
    ...
}

public byte[] getData(String path, Stat stat, Watcher watcher) throws NoNodeException {
    return this.dataTree.getData(path, stat, watcher);
}

　　紧接着，将数据节点路径和ServerCnxn对象存储在WatcherManager的watchTable和watch2Paths中。前者是路径维度，后者是Watcher维度

public byte[] getData(String path, Stat stat, Watcher watcher) throws NoNodeException {
    DataNode n = (DataNode)this.nodes.get(path);
    if (n == null) {
        throw new NoNodeException();
    } else {
        synchronized(n) {
            n.copyStat(stat);
            if (watcher != null) {
                // 添加watcher
                this.dataWatches.addWatch(path, watcher);
            }

            return n.data;
        }
    }
}

public synchronized void addWatch(String path, Watcher watcher) {
    HashSet<Watcher> list = (HashSet)this.watchTable.get(path);
    if (list == null) {
        list = new HashSet(4);
        this.watchTable.put(path, list);
    }

    list.add(watcher);
    HashSet<String> paths = (HashSet)this.watch2Paths.get(watcher);
    if (paths == null) {
        paths = new HashSet();
        this.watch2Paths.put(watcher, paths);
    }

    paths.add(path);
}

// 路径维度
private final HashMap<String, HashSet<Watcher>> watchTable = new HashMap();
// Watcher维度
private final HashMap<Watcher, HashSet<String>> watch2Paths = new HashMap();

　　当服务端处理完毕之后，客户端的SendThread线程负责接收服务端的响应，finishPacket方法会从packet中取出WatchRegistration并注册到ZKWatchManager中：

private void finishPacket(ClientCnxn.Packet p) {
    // 客户端注册wathcer
    if (p.watchRegistration != null) {
        p.watchRegistration.register(p.replyHeader.getErr());
    }

    if (p.cb == null) {
        synchronized(p) {
            p.finished = true;
            p.notifyAll();
        }
    } else {
        p.finished = true;
        this.eventThread.queuePacket(p);
    }

}

public void register(int rc) {
    if (this.shouldAddWatch(rc)) {
        Map<String, Set<Watcher>> watches = this.getWatches(rc);
        synchronized(watches) {
            // 根据路径拿到
            Set<Watcher> watchers = (Set)watches.get(this.clientPath);
            if (watchers == null) {
                watchers = new HashSet();
                watches.put(this.clientPath, watchers);
            }

            ((Set)watchers).add(this.watcher);
        }
    }

}

 

通知

 　　当服务端对应的数据节点内容发生改变，那么会触发watcher，对应的代码在org.apache.zookeeper.server.DataTree#setData

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;
        // 赋值node
        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;
        if ((lastPrefix = this.getMaxPrefixWithQuota(path)) != null) {
            this.updateBytes(lastPrefix, (long)((data == null ? 0 : data.length) - (lastdata == null ? 0 : lastdata.length)));
        }

        // 触发watcher
        this.dataWatches.triggerWatch(path, EventType.NodeDataChanged);
        return s;
    }
}

　　触发watcher，从watchTable和watch2Paths中移除该路径的watcher。这里可以看出，Watcher在服务端是一次性的，触发一次就失效了

public Set<Watcher> triggerWatch(String path, EventType type) {
    return this.triggerWatch(path, type, (Set)null);
}

public Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {
    WatchedEvent e = new WatchedEvent(type, KeeperState.SyncConnected, path);
    HashSet watchers;

    // 这个同步代码块主要做的就是从watchTable和watch2Paths中移除该路径的watcher
    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));

        // watcher调用，这里的e对象里只有通知状态（KeeperState）、事件类型（EventType）以及节点路径（Path）
        // 没有修改过后的新值也没有老的值
        w.process(e);
    }
}

　　最后看一下process方法里，其实做的事情就是把事件发送给客户端，所以我们可以看出，真正的回调和业务逻辑执行都在客户端org.apache.zookeeper.server.NIOServerCnxn#process：

public synchronized void process(WatchedEvent event) {
 　　// 请求头标记-1，表明是通知
    ReplyHeader h = new ReplyHeader(-1, -1L, 0);
    if (LOG.isTraceEnabled()) {
        ZooTrace.logTraceMessage(LOG, 64L, "Deliver event " + event + " to 0x" + Long.toHexString(this.sessionId) + " through " + this);
    }

    WatcherEvent e = event.getWrapper();
    // 发送通知
    this.sendResponse(h, e, "notification");
}

　　客户端收到该通知，由org.apache.zookeeper.ClientCnxn.SendThread#readResponse处理，主要做的就是反序列化然后交给EventThread线程

void readResponse(ByteBuffer incomingBuffer) throws IOException {
    ...
    // 如果是通知
    } 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));
        }

        // 交给EventThread线程处理
        ClientCnxn.this.eventThread.queueEvent(we);
    } 
    ...
}

　　然后从之前注册的ZKWatcherManager中获取到所有该路径的watcher，注意了，客户端的Watcher机制也是一次性的！

public void queueEvent(WatchedEvent event) {
    if (event.getType() != EventType.None || this.sessionState != event.getState()) {
        this.sessionState = event.getState();
        ClientCnxn.WatcherSetEventPair pair = new ClientCnxn.WatcherSetEventPair(ClientCnxn.this.watcher.materialize(event.getState(), event.getType(), event.getPath()), event);
        this.waitingEvents.add(pair);
    }
}


public Set<Watcher> materialize(KeeperState state, EventType type, String clientPath) {
    ...
    // 把该路径下的所有Watcher都拿出来
    // remove方法，所以客户端也是一次性的，一旦触发，watcher就失效了
    case NodeDataChanged:
    case NodeCreated:
        var6 = this.dataWatches;
        synchronized(this.dataWatches) {
            this.addTo((Set)this.dataWatches.remove(clientPath), result);
        }

        var6 = this.existWatches;
        synchronized(this.existWatches) {
            this.addTo((Set)this.existWatches.remove(clientPath), result);
            break;
        }
    ...
}

　　最后EventThread会从waitingEvents队列中取出Watcher并执行串行化同步处理。看一下这个方法：org.apache.zookeeper.ClientCnxn.EventThread#processEvent

private void processEvent(Object event) {
    try {
        if (event instanceof ClientCnxn.WatcherSetEventPair) {
             ClientCnxn.WatcherSetEventPair pair = (ClientCnxn.WatcherSetEventPair)event;
             Iterator i$ = pair.watchers.iterator();
  
             while(i$.hasNext()) {
             
                 // 这里的watcher就是客户端传入的watcher，里面有真正的回调逻辑代码
                Watcher watcher = (Watcher)i$.next();
 
                try {
                    watcher.process(pair.event);
                } catch (Throwable var7) {
                    ClientCnxn.LOG.error("Error while calling watcher ", var7);
                }
             }
        } else {
        ...
    }
    ...
}

　　嗯，就是这样，走完了，从网上找到一张图，我觉得画的很不错。以上三步骤，注册，存储，通知可以结合这张图来看，最好请打开原图来看：

三.总结

Watcher特性总结

一次性

　　无论客户端还是服务端，一旦watcher被触发，都会被移除

客户端串行执行

　　从源码也看到了，watcher回调是串行同步化执行过程，注意不要一个watcher中放很多处理逻辑造成影响别的watcher回调

性能轻量

　　注册watcher把watcher对象传给服务端，回调的时候并不会告诉节点的具体变化前后的内容。非常轻量　　

时效

　　发生CONNECTIONLOSS之后，只要在session_timeout之内再次连接上（即不发生SESSIONEXPIRED），那么这个连接注册的watches依然在。

节点通知

　　guava to java is Curator to ZooKeeper，开源客户端Curator引入Cache实现对服务端事件的监听，从而大大简化了原生API开发的繁琐过程。

　　虽然我们可以通过Curator或者ZKClient避免每次要watcher注册的痛苦，但是我们无法保证在节点更新频率很高的情况下客户端能收到每一次节点变化的通知

　　原因在于：当一次数据修改，通知客户端，客户端再次注册watch，在这个过程中，可能数据已经发生了许多次数据修改

参考：

偷来的图：https://blog.csdn.net/huyangyamin/article/details/77743624