Zookeeper-watcher机制源码分析

Posted 2020-12-25 xuespring

Watcher的基本流程

ZooKeeper 的 Watcher 机制，总的来说可以分为三个过程：客户端注册 Watcher、服务器处理 Watcher 和客户端回调 Watcher

客户端注册watcher有3种方式，getData、exists、getChildren；以如下代码为例来分析整个触发机制的原理

ZooKeeper zookeeper=new ZooKeeper(“192.168.11.152:2181”,4000,new Watcher(){ public void processor(WatchedEvent event){  System.out.println(“event.type”); } })；   zookeeper.create(“/mic”,”0”.getByte(),ZooDefs.Ids. OPEN_ACL_UNSAFE,CreateModel. PERSISTENT); //创建节点   zookeeper.exists(“/mic”,true); //注册监听   zookeeper.setData(“/mic”, “1”.getByte(),-1) ; //修改节点的值触发监听

 

ZooKeeper API的初始化过程

ZooKeeper zookeeper=new ZooKeeper(“192.168.11.152:2181”,4000,new Watcher(){ public void processor(WatchedEvent event){  System.out.println(“event.type”); } })；

在创建一个 ZooKeeper 客户端对象实例时，我们通过new Watcher()向构造方法中传入一个默认的 Watcher, 这个 Watcher 将作为整个 ZooKeeper会话期间的默认 Watcher，会一直被保存在客户端 ZKWatchManager 的 defaultWatcher 中;代码如下

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,             boolean canBeReadOnly, HostProvider aHostProvider,             ZKClientConfig clientConfig) throws IOException {         LOG.info("Initiating client connection, connectString=" + connectString                 + " sessionTimeout=" + sessionTimeout + " watcher=" + watcher);           if (clientConfig == null) {             clientConfig = new ZKClientConfig();         }         this.clientConfig = clientConfig;         watchManager = defaultWatchManager();         watchManager.defaultWatcher = watcher;  --在这里将watcher设置到ZKWatchManager         ConnectStringParser connectStringParser = new ConnectStringParser(                 connectString);         hostProvider = aHostProvider;         --初始化了ClientCnxn，并且调用cnxn.start()方法         cnxn = new ClientCnxn(connectStringParser.getChrootPath(),                 hostProvider, sessionTimeout, this, watchManager,                 getClientCnxnSocket(), canBeReadOnly);         cnxn.start();     }

ClientCnxn:是Zookeeper客户端和Zookeeper服务器端进行通信和事件通知处理的主要类，它内部包含两个类，

1. SendThread  ：负责客户端和服务器端的数据通信, 也包括事件信息的传输

2. EventThread :  主要在客户端回调注册的Watchers进行通知处理

ClientCnxn初始化

public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,             ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,             long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {         this.zooKeeper = zooKeeper;         this.watcher = watcher;         this.sessionId = sessionId;         this.sessionPasswd = sessionPasswd;         this.sessionTimeout = sessionTimeout;         this.hostProvider = hostProvider;         this.chrootPath = chrootPath;           connectTimeout = sessionTimeout / hostProvider.size();         readTimeout = sessionTimeout * 2 / 3;         readOnly = canBeReadOnly;           sendThread = new SendThread(clientCnxnSocket);  --初始化sendThread         eventThread = new EventThread();                --初始化eventThread         this.clientConfig=zooKeeper.getClientConfig();     }       public void start() { --启动两个线程         sendThread.start();         eventThread.start();     }

 

客户端通过exists注册监听

zookeeper.exists(“/mic”,true); //注册监听

通过exists方法来注册监听，代码如下

public Stat exists(final String path, Watcher watcher)         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 ExistsWatchRegistration(watcher, clientPath); //构建ExistWatchRegistration         }           final String serverPath = prependChroot(clientPath);           RequestHeader h = new RequestHeader();         h.setType(ZooDefs.OpCode.exists);  //设置操作类型为exists         ExistsRequest request = new ExistsRequest();  // 构造ExistsRequest         request.setPath(serverPath);         request.setWatch(watcher != null);  //是否注册监听         SetDataResponse response = new SetDataResponse();  //设置服务端响应的接收类 //将封装的RequestHeader、ExistsRequest、SetDataResponse、WatchRegistration添加到发送队列         ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);          if (r.getErr() != 0) {             if (r.getErr() == KeeperException.Code.NONODE.intValue()) {                 return null;             }             throw KeeperException.create(KeeperException.Code.get(r.getErr()),                     clientPath);         }         //返回exists得到的结果（Stat信息）         return response.getStat().getCzxid() == -1 ? null : response.getStat();     }

cnxn.submitRequest

public ReplyHeader submitRequest(RequestHeader h, Record request,             Record response, WatchRegistration watchRegistration,             WatchDeregistration watchDeregistration)             throws InterruptedException {         ReplyHeader r = new ReplyHeader();         //将消息添加到队列,并构造一个Packet传输对象         Packet packet = queuePacket(h, r, request, response, null, null, null, null, watchRegistration, watchDeregistration);         synchronized (packet) {             while (!packet.finished) { //在数据包没有处理完成之前，一直阻塞                 packet.wait();             }         }         return r;     }

 

public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,             Record response, AsyncCallback cb, String clientPath,             String serverPath, Object ctx, WatchRegistration watchRegistration,             WatchDeregistration watchDeregistration) {         //将相关传输对象转化成Packet         Packet packet = null;         packet = new Packet(h, r, request, response, watchRegistration);         packet.cb = cb;         packet.ctx = ctx;         packet.clientPath = clientPath;         packet.serverPath = serverPath;         packet.watchDeregistration = watchDeregistration;                  synchronized (state) {             if (!state.isAlive() || closing) {                 conLossPacket(packet);             } else {                 if (h.getType() == OpCode.closeSession) {                     closing = true;                 }                 outgoingQueue.add(packet); //添加到outgoingQueue             }         }         sendThread.getClientCnxnSocket().packetAdded();//此处是多路复用机制，唤醒Selector，告诉他有数据包添加过来了         return packet;     }

在 ZooKeeper 中，Packet 是一个最小的通信协议单元，即数据包。Pakcet 用于进行客户端与服务端之间的网络传输，任何需要传输的对象都需要包装成一个 Packet 对象。在 ClientCnxn 中 WatchRegistration 也会被封装到 Pakcet 中，然后由 SendThread 线程调用queuePacket方法把 Packet 放入发送队列中等待客户端发送，这又是一个异步过程，分布式系统采用异步通信是一个非常常见的手段

SendThread的发送过程

在初始化连接的时候，zookeeper初始化了两个线程并且启动了。接下来我们来分析SendThread的发送过程，因为是一个线程，所以启动的时候会调用SendThread.run方法

public void run() {             clientCnxnSocket.introduce(this, sessionId, outgoingQueue);             clientCnxnSocket.updateNow();             clientCnxnSocket.updateLastSendAndHeard();             int to;             long lastPingRwServer = Time.currentElapsedTime();             final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds             while (state.isAlive()) {                 try {                     if (!clientCnxnSocket.isConnected()) {// 如果没有连接：发起连接                         // don‘t re-establish connection if we are closing                         if (closing) {                             break;                         }                         startConnect(); //发起连接                         clientCnxnSocket.updateLastSendAndHeard();                     }                       if (state.isConnected()) { //如果是连接状态，则处理sasl的认证授权                         // determine whether we need to send an AuthFailed event.                         if (zooKeeperSaslClient != null) {                             boolean sendAuthEvent = false;                             if (zooKeeperSaslClient.getSaslState() == ZooKeeperSaslClient.SaslState.INITIAL) {                                 try {                                     zooKeeperSaslClient.initialize(ClientCnxn.this);                                 } catch (SaslException e) {                                    LOG.error("SASL authentication with Zookeeper Quorum member failed: " + e);                                     state = States.AUTH_FAILED;                                     sendAuthEvent = true;                                 }                             }                             KeeperState authState = zooKeeperSaslClient.getKeeperState();                             if (authState != null) {                                 if (authState == KeeperState.AuthFailed) {                                     // An authentication error occurred during authentication with the Zookeeper Server.                                     state = States.AUTH_FAILED;                                     sendAuthEvent = true;                                 } else {                                     if (authState == KeeperState.SaslAuthenticated) {                                         sendAuthEvent = true;                                     }                                 }                             }                               if (sendAuthEvent == true) {                                 eventThread.queueEvent(new WatchedEvent(                                       Watcher.Event.EventType.None,                                       authState,null));                             }                         }                         to = readTimeout - clientCnxnSocket.getIdleRecv();                     } else {                         to = connectTimeout - clientCnxnSocket.getIdleRecv();                     }                     //to,表示客户端距离timeout还剩多少时间，准备发起ping连接                     if (to <= 0) {//表示已经超时了。                         String warnInfo;                         warnInfo = "Client session timed out, have not heard from server in "                             + clientCnxnSocket.getIdleRecv()                             + "ms"                             + " for sessionid 0x"                             + Long.toHexString(sessionId);                         LOG.warn(warnInfo);                         throw new SessionTimeoutException(warnInfo);                     }                     if (state.isConnected()) {                         //计算下一次ping请求的时间                         int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() -                          ((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);                         //send a ping request either time is due or no packet sent out within MAX_SEND_PING_INTERVAL                         if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) {                             sendPing(); //发送ping请求                             clientCnxnSocket.updateLastSend();                         } else {                             if (timeToNextPing < to) {                                 to = timeToNextPing;                             }                         }                     }                     // If we are in read-only mode, seek for read/write server                     if (state == States.CONNECTEDREADONLY) {                         long now = Time.currentElapsedTime();                         int idlePingRwServer = (int) (now - lastPingRwServer);                         if (idlePingRwServer >= pingRwTimeout) {                             lastPingRwServer = now;                             idlePingRwServer = 0;                             pingRwTimeout =                                 Math.min(2*pingRwTimeout, maxPingRwTimeout);                             pingRwServer();                         }                         to = Math.min(to, pingRwTimeout - idlePingRwServer);                     }                     调用clientCnxnSocket，发起传输                     其中 pendingQueue是一个用来存放已经发送、等待回应的Packet队列， clientCnxnSocket默认使用ClientCnxnSocketNIO（ps：还记得在哪里初始化吗？在实例化zookeeper的时候）                     clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);                 } catch (Throwable e) {                     if (closing) {                         if (LOG.isDebugEnabled()) {                             // closing so this is expected                             LOG.debug("An exception was thrown while closing send thread for session 0x"                                     + Long.toHexString(getSessionId())                                     + " : " + e.getMessage());                         }                         break;                     } else {                         // this is ugly, you have a better way speak up                         if (e instanceof SessionExpiredException) {                             LOG.info(e.getMessage() + ", closing socket connection");                         } else if (e instanceof SessionTimeoutException) {                             LOG.info(e.getMessage() + RETRY_CONN_MSG);                         } else if (e instanceof EndOfStreamException) {                             LOG.info(e.getMessage() + RETRY_CONN_MSG);                         } else if (e instanceof RWServerFoundException) {                             LOG.info(e.getMessage());                         } else {                             LOG.warn(                                     "Session 0x"                                             + Long.toHexString(getSessionId())                                             + " for server "                                             + clientCnxnSocket.getRemoteSocketAddress()                                             + ", unexpected error"                                             + RETRY_CONN_MSG, e);                         }                         // At this point, there might still be new packets appended to outgoingQueue.                         // they will be handled in next connection or cleared up if closed.                         cleanup();                         if (state.isAlive()) {                             eventThread.queueEvent(new WatchedEvent(                                     Event.EventType.None,                                     Event.KeeperState.Disconnected,                                     null));                         }                         clientCnxnSocket.updateNow();                         clientCnxnSocket.updateLastSendAndHeard();                     }                 }             }             synchronized (state) {                 // When it comes to this point, it guarantees that later queued                 // packet to outgoingQueue will be notified of death.                 cleanup();             }             clientCnxnSocket.close();             if (state.isAlive()) {                 eventThread.queueEvent(new WatchedEvent(Event.EventType.None,                         Event.KeeperState.Disconnected, null));             }             ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(),                     "SendThread exited loop for session: 0x"                            + Long.toHexString(getSessionId()));         }

client 和 server的网络交互

@Override     void doTransport(int waitTimeOut, List<Packet> pendingQueue, ClientCnxn cnxn) throws IOException, InterruptedException {         try {             if (!firstConnect.await(waitTimeOut, TimeUnit.MILLISECONDS)) {                 return;             }             Packet head = null;             if (needSasl.get()) {                 if (!waitSasl.tryAcquire(waitTimeOut, TimeUnit.MILLISECONDS)) {                     return;                 }             } else {                   //判断outgoingQueue是否存在待发送的数据包，不存在则直接返回                 if ((head = outgoingQueue.poll(waitTimeOut, TimeUnit.MILLISECONDS)) == null) {                     return;                 }             }             // check if being waken up on closing.             if (!sendThread.getZkState().isAlive()) {                 // adding back the patck to notify of failure in conLossPacket().                 addBack(head);                 return;             }             // channel disconnection happened             if (disconnected.get()) { //异常流程，channel关闭了，讲当前的packet添加到addBack中                 addBack(head);                 throw new EndOfStreamException("channel for sessionid 0x"                         + Long.toHexString(sessionId)                         + " is lost");             }             if (head != null) { //如果当前存在需要发送的数据包，则调用doWrite方法，pendingQueue表示处于已经发送过等待响应的packet队列                 doWrite(pendingQueue, head, cnxn);             }         } finally {             updateNow();         }     }

DoWrite方法

private void doWrite(List<Packet> pendingQueue, Packet p, ClientCnxn cnxn) {         updateNow();         while (true) {             if (p != WakeupPacket.getInstance()) {                 if ((p.requestHeader != null) && //判断请求头以及判断当前请求类型不是ping或者auth操作                         (p.requestHeader.getType() != ZooDefs.OpCode.ping) &&                         (p.requestHeader.getType() != ZooDefs.OpCode.auth)) {                     p.requestHeader.setXid(cnxn.getXid());  //设置xid，这个xid用来区分请求类型                     synchronized (pendingQueue) {                         pendingQueue.add(p); //将当前的packet添加到pendingQueue队列中                     }                 }                 sendPkt(p); //将数据包发送出去             }             if (outgoingQueue.isEmpty()) {                 break;             }             p = outgoingQueue.remove();         }     }

sendPkt

private void sendPkt(Packet p) {         // Assuming the packet will be sent out successfully. Because if it fails,         // the channel will close and clean up queues.         p.createBB(); //序列化请求数据         updateLastSend(); //更新最后一次发送updateLastSend         sentCount++; //更新发送次数         channel.write(ChannelBuffers.wrappedBuffer(p.bb)); //通过nio channel发送字节缓存到服务端     }

 

createBB

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) {                     requestHeader.serialize(boa, "header"); //序列化header头(requestHeader)                 }                 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"); //序列化request(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);             }         }

从createBB方法中，我们看到在底层实际的网络传输序列化中，zookeeper只会讲requestHeader和request两个属性进行序列化，即只有这两个会被序列化到底层字节数组中去进行网络传输，不会将watchRegistration相关的信息进行网络传输。

总结

用户调用exists注册监听以后，会做几个事情

1. 讲请求数据封装为packet，添加到outgoingQueue
2. SendThread这个线程会执行数据发送操作，主要是将outgoingQueue队列中的数据发送到服务端
3. 通过clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this); 其中ClientCnxnSocket只zookeeper客户端和服务端的连接通信的封装，有两个具体的实现类ClientCnxnSocketNetty和ClientCnxnSocketNIO;具体使用哪一个类来实现发送，是在初始化过程是在实例化Zookeeper的时候设置的，代码如下

 

cnxn = new ClientCnxn(connectStringParser.getChrootPath(),                 hostProvider, sessionTimeout, this, watchManager,                 getClientCnxnSocket(), canBeReadOnly);

private ClientCnxnSocket getClientCnxnSocket() throws IOException {         String clientCnxnSocketName = getClientConfig().getProperty(                 ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET);         if (clientCnxnSocketName == null) {             clientCnxnSocketName = ClientCnxnSocketNIO.class.getName();         }         try {             Constructor<?> clientCxnConstructor = Class.forName(clientCnxnSocketName).getDeclaredConstructor(ZKClientConfig.class);             ClientCnxnSocket clientCxnSocket = (ClientCnxnSocket) clientCxnConstructor.newInstance(getClientConfig());             return clientCxnSocket;         } catch (Exception e) {             IOException ioe = new IOException("Couldn‘t instantiate "                     + clientCnxnSocketName);             ioe.initCause(e);             throw ioe;         }     }

4.基于第3步，最终会在ClientCnxnSocketNetty方法中执行sendPkt将请求的数据包发送到服务端

对Java技术，架构技术感兴趣的同学，欢迎加QQ群619881427，一起学习，相互讨论。

群内已经有小伙伴将知识体系整理好（源码，笔记，PPT，学习视频），欢迎加群免费领取。

分享给喜欢Java，喜欢编程，有梦想成为架构师的程序员们，希望能够帮助到你们。

不是Java程序员也没关系，帮忙转发给更多朋友！谢谢。

分享一个小技巧点击阅读原文也可以轻松获取到学习资料哦！！