zookeeper 监听器原理源码分析学习

妖狐艹你老母 2022-03-29 13:46 197阅读 0赞

##### 监听器原理 #####

1.客户端有一个主线程，
    2.主线程创建一个eventThread()和sendThread()
    3.一个负责监听，一个负责网络连接
    4.通过connect线程将注册的监听事件发送给zookeeper，
    5.在zookeeper的注册监听器列表中将注册的监听事件添加到列表中
    6.zookeeper监听有数据或路径发生变化，就将这个消息发送给监听线程
    7.listenner线程调用Process()方法
    常见监听：监听节点数据的变化
              监听子节点增减的变化

##### zookeeper cli类型构造器 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70]

> 创建两个线程sendThread和eventThread  
> sentThread 发送信息的线程（负责网络连接通信）  
> eventThread发生改变进行通知的线程(负责监听)

##### 启动两个线程 #####

![在这里插入图片描述][20190108184508142.png]

##### EventThread线程run()方法 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 1]

@Override
            public void run() {
               try {
                  isRunning = true; //boolean 标记为
                  while (true) { //死循环
                     Object event = waitingEvents.take(); // waitingEvents是一个LinkedBlockingQueue队列， take()方法从队列中拿取数据,通过
                     if (event == eventOfDeath) {
                        wasKilled = true;
                     } else {
                        processEvent(event); //processEvent,处理发生在组件内的事件,应该是发送消息给监听者
                     }
                     if (wasKilled)
                        synchronized (waitingEvents) {
                           if (waitingEvents.isEmpty()) { //如果waitingEvents为空，进行等待，waitingEvents就是那个LinkedBlockingQueue队列
                              isRunning = false;
                              break;
                           }
                        }
                  }
               } catch (InterruptedException e) {
                  LOG.error("Event thread exiting due to interruption", e); //log相关
               }
    
                LOG.info("EventThread shut down for session: 0x{}",
                         Long.toHexString(getSessionId()));
            }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 2]

> take()方法中调用dequeue()方法，从LinkedBlockingQueue队列中，拿走数据

/**
         * Removes a node from head of queue.
         *
         * @return the node
         */
        private E dequeue() {  
            // assert takeLock.isHeldByCurrentThread();
            // assert head.item == null;
            Node<E> h = head; //获取头部节点
            Node<E> first = h.next; //改变头结点
            h.next = h; // help GC
            head = first;
            E x = first.item;
            first.item = null;
            return x;
        }

##### sendThread线程run()方法 #####

@Override
            public void run() {
                clientCnxnSocket.introduce(this,sessionId);
                clientCnxnSocket.updateNow();
                clientCnxnSocket.updateLastSendAndHeard();
                int to;
                long lastPingRwServer = System.currentTimeMillis();
                final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds  //最大发送心跳间隔
                while (state.isAlive()) {
                    try {
                        if (!clientCnxnSocket.isConnected()) { //如果未连接，进行休眠
                            if(!isFirstConnect){
                                try {
                                    Thread.sleep(r.nextInt(1000)); //随机睡眠1-1000的数字 (ms)
                                } catch (InterruptedException e) {
                                    LOG.warn("Unexpected exception", e);
                                }
                            }
                            // don't re-establish connection if we are closing //如果关闭状态不进行建立连接
                            if (closing || !state.isAlive()) {
                                break;
                            }
                            startConnect();  //通过clientCnxnSocket进行通信连接
                            clientCnxnSocket.updateLastSendAndHeard();//
                        }
    					//如果建立连接
                        if (state.isConnected()) {
                            // 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;
                                        }
                                    }
                                }
    							//SendThread负责发送outgoingQueue中的请求
                                if (sendAuthEvent == true) {
                                    eventThread.queueEvent(new WatchedEvent( //quequEvent()中将pair添加到waitingEvents中
                                          Watcher.Event.EventType.None,
                                          authState,null));
                                }
                            }
                            to = readTimeout - clientCnxnSocket.getIdleRecv();
                        } else {
                            to = connectTimeout - clientCnxnSocket.getIdleRecv();
                        }
                        
                        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()) {
                        	//1000(1 second) is to prevent race condition missing to send the second ping
                        	//also make sure not to send too many pings when readTimeout is small 
                            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(); //
                                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 = System.currentTimeMillis();
                            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);
                        }
    					//doIO(pendingQueue, outgoingQueue, cnxn)发送数据
                        clientCnxnSocket.doTransport(to, pendingQueue, outgoingQueue, 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);
                            }
                            cleanup();
                            if (state.isAlive()) {
                                eventThread.queueEvent(new WatchedEvent(
                                        Event.EventType.None,
                                        Event.KeeperState.Disconnected,
                                        null));
                            }
                            clientCnxnSocket.updateNow();
                            clientCnxnSocket.updateLastSendAndHeard();
                        }
                    }
                }
                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()));
            }

> ClientCnxn的后台线程，以及接受Server端返回的数据，还包括心跳的发送，run()  
> 先看主要IO流程  
> clientCnxnSocket.doTransport–>doIO(pendingQueue, outgoingQueue, cnxn)–>sockKey.isWritable()–>sock.write([p.bb][])–>sockKey.isReadable()–> sendThread.readResponse(incomingBuffer)–> finishPacket(packet)–>p.notifyAll();  
> 这里在sock.write([p.bb][])，hui将Packet加入到pendingQueue队列  
> 如果是读取完了之后，自然要从pendingQueue移除remove

//readResponse()函数调用finishPacket()
        private void finishPacket(Packet p) {
            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;
                eventThread.queuePacket(p);
            }
        }

> 相关源码后期还会继续去理解，补充…

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70]: /images/20220329/c107c92795974774bb08525e2f6f80bf.png
[20190108184508142.png]: /images/20220329/e0e37e0c7dd2474da4a0fd2f26f751e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 1]: /images/20220329/e072f4e2ad4440a382a1bac3b40516a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 2]: /images/20220329/93ae8d9512e14b29bbd37a018dbc607d.png
[p.bb]: http://p.bb