public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, HostProvider aHostProvider, ZKClientConfig clientConfig) throws IOException {
   if (clientConfig == null) {
       clientConfig = new ZKClientConfig();
   }
   this.clientConfig = clientConfig;
   // 观察者管理器
   watchManager = defaultWatchManager();
   watchManager.defaultWatcher = watcher;
   ConnectStringParser connectStringParser = new ConnectStringParser(
           connectString);
   hostProvider = aHostProvider;

   // ClientCnxn: client的通讯组件，负责socket通讯（发送请求等，且支持透明地切换到不同的Server）
   // ClientCnxnSocket : 真正通讯的组件，ClientCnxnSocketNIO、ClientCnxnSocketNetty
   cnxn = new ClientCnxn(connectStringParser.getChrootPath(),// 客户端隔离命名空间，相当于试图模式。只对某个path的进行操作
           hostProvider,// 服务器提供者列表
           sessionTimeout,//会话超时时间
           this,
           watchManager, // 监听管理器
           getClientCnxnSocket(),// 客户端套接字通讯工具 ClientCnxnSocket 比如 ClientCnxnSocketNetty
           canBeReadOnly);
  // 初始化
   cnxn.start();
}

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);
   // 事件线程，当注册的节点发生变化时，server端会通知client，EventThread用来获取通知，并触发相应的Watcher
   eventThread = new EventThread();
   this.clientConfig = zooKeeper.getClientConfig();
}

public void start() {
   sendThread.start();
   eventThread.start();
}

public void run() {
  // 引入 outgoingQueue
  clientCnxnSocket.introduce(this, sessionId, outgoingQueue);
  clientCnxnSocket.updateNow();
  clientCnxnSocket.updateLastSendAndHeard();
  int to;
  long lastPingRwServer = Time.currentElapsedTime();
  // 10 秒发一次 Ping 操作
  final int MAX_SEND_PING_INTERVAL = 10000;
  InetSocketAddress serverAddress = null;
  while (state.isAlive()) {
      // 链接断开时，重连
      if (!clientCnxnSocket.isConnected()) {
          if (closing) {
              break;
          }
          if (rwServerAddress != null) {
              serverAddress = rwServerAddress;
              rwServerAddress = null;
          } else {
              //如果所有的host都试了一遍，则停顿 1s
              serverAddress = hostProvider.next(1000);
          }
          // 链接服务器，首次连接时同时
          startConnect(serverAddress);
          clientCnxnSocket.updateLastSendAndHeard();
      }
      // ...
      if (state.isConnected()) {
          int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() -
                  ((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);
          if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) {
              //当有 MAX_SEND_PING_INTERVAL秒没发送请求，则发送 ping 心跳
              sendPing();
              clientCnxnSocket.updateLastSend();
          } else {
              if (timeToNextPing < to) {
                  to = timeToNextPing;
              }
          }
      }
      //对于只读的则重试ping到读写服务上
      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 从 outgoingQueue中取请求发送
      clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);
  }
  // ...
}

 public R execute() {
   try {
       return queue().get();
   } catch (Exception e) {
       throw Exceptions.sneakyThrow(decomposeException(e));
   }
}
public Future<R> queue() {  
   //第一步执行
   final Future<R> delegate = **toObservable()**.toBlocking().toFuture();
}

//消息发送者
Observable<String> sender = Observable.create(new Observable.OnSubscribe<String>() {
 @Override
  public void call(Subscriber<? super String> subscriber) {
      while(下载进度<100%) {
        subscriber.onNext("当前已下载" + 下载进度);
      }

      subscriber.onNext("Hello World");  
  }
});
//消息订阅者
Observer<String> receiver = new Observer<String>() {
  @Override
  public void onCompleted() {
      //数据接收完成时调用
  }
  @Override
  public void onError(Throwable e) {
      //发生错误调用
  }
  @Override
  public void onNext(String s) {
    // 获取页面句柄，渲染后展示给用户看当前下载进度
    viewHander.show(s);
  }
};
//消息订阅
sender.subscribe(receiver);