Zookeeper 3.4.6 Client端流程粗略梳理

devil20

　　clientCnxnSocket.introduce(this,sessionId);
　　clientCnxnSocket.updateNow();
　　clientCnxnSocket.updateLastSendAndHeard();
　　int to;
　　long lastPingRwServer = System.currentTimeMillis();
　　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;
　　}
　　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 = 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.");