zookeeper心跳机制流程梳理

mingche

@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));
　　} catch (InterruptedException e) {
　　LOG.warn("Unexpected exception", e);
　　}
　　}
　　// don't re-establish connection if we are closing
　　if (closing || !state.isAlive()) {
　　break;
　　}
　　//注册OP_CONNECT事件      sockKey = sock.register(selector, SelectionKey.OP_CONNECT);
　　startConnect();
　　//最近一次发送接收时间
　　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;
　　}
　　}
　　}
　　
　　if (sendAuthEvent == true) {
　　eventThread.queueEvent(new WatchedEvent(
　　Watcher.Event.EventType.None,
　　authState,null));
　　}
　　}
　　//读超时-读空闲时间
　　to = readTimeout - clientCnxnSocket.getIdleRecv();
　　} else {
　　//连接超时-读空闲时间
　　to = connectTimeout - clientCnxnSocket.getIdleRecv();
　　}
　　
　　if (to  1000) ? 1000 : 0);
　　//send a ping request either time is due or no packet sent out within MAX_SEND_PING_INTERVAL
　　//写空闲的时候才发送心跳
　　if (timeToNextPing  MAX_SEND_PING_INTERVAL) {
　　/**
　　* 发送ping请求，
　　* 首次通过queuePacket将请求加入outgoingQueue队列
　　* 唤醒sendThread.getClientCnxnSocket().wakeupCnxn();-->selector.wakeup();
　　*/
　　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>　　if (idlePingRwServer >= pingRwTimeout) {
　　lastPingRwServer = now;
　　idlePingRwServer = 0;
　　pingRwTimeout =
　　Math.min(2*pingRwTimeout, maxPingRwTimeout);
　　pingRwServer();
　　}

　　to = Math.min(to, pingRwTimeout ->　　}
　　
　　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 exitedloop.");
　　}