ZooKeeper源码阅读系列-zk-client解析

ZOOBIN="${BASH_SOURCE-$0}"
        ZOOBIN="$(dirname "${ZOOBIN}")"
        ZOOBINDIR="$(cd "${ZOOBIN}"; pwd)"
        if [ -e "$ZOOBIN/../libexec/zkEnv.sh" ]; then
        . "$ZOOBINDIR"/../libexec/zkEnv.sh
        else
        . "$ZOOBINDIR"/zkEnv.sh
        fi
        "$JAVA" "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" \
        -cp "$CLASSPATH" $CLIENT_JVMFLAGS $JVMFLAGS
        org.apache.zookeeper.ZooKeeperMain "$@"
复制代码

void doIO(List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn cnxn)
     throws InterruptedException, IOException {
  SocketChannel sock = (SocketChannel) sockKey.channel();//获取Channel
  if (sock == null) {
     throw new IOException("Socket is null!");
  }
  if (sockKey.isReadable()) {//读就绪
     int rc = sock.read(incomingBuffer);//读取长度
     if (rc < 0) { ///如果<0,表示读到末尾了,eof了异常关闭了
        throw new EndOfStreamException(
              "Unable to read additional data from server sessionid 0x"
                    + Long.toHexString(sessionId)
                    + ", likely server has closed socket");
     }
     if (!incomingBuffer.hasRemaining()) {//如果还有数据
        incomingBuffer.flip(); //切换模式
        if (incomingBuffer == lenBuffer) {
           recvCount++;//新增接收次数
           readLength();//获取len并给incomingBuffer分配对应空间
        } else if (!initialized) {//如果连接还未初始化
           readConnectResult();//读取connect并回复
           enableRead();//启用读
           if (findSendablePacket(outgoingQueue,cnxn.sendThread.clientTunneledAuthenticationInProgress()) != null) {
              //如果有可以发送的packet
              enableWrite();//允许写，因为有要发送的packet
           }
           lenBuffer.clear();
           incomingBuffer = lenBuffer;
           updateLastHeard();
           initialized = true;//初始化完成
        } else {//如果已连接，并且已经给incomingBuffer分配了对应len的空间
           sendThread.readResponse(incomingBuffer);//读取response
           lenBuffer.clear();
           incomingBuffer = lenBuffer;
           updateLastHeard();//更新上次接收时间
        }
     }
  }
  //如果是写事件
  if (sockKey.isWritable()) {
     synchronized(outgoingQueue) {//锁住队列
        Packet p = findSendablePacket(outgoingQueue, cnxn.sendThread.clientTunneledAuthenticationInProgress());//找到可以发送的packet
        if (p != null) {
           updateLastSend();
           if (p.bb == null) {//packet内部还未生成bytebuffer
              if ((p.requestHeader != null) &&
                    (p.requestHeader.getType() != OpCode.ping) &&
                    (p.requestHeader.getType() != OpCode.auth)) {
                 p.requestHeader.setXid(cnxn.getXid());
              }
              p.createBB();
           }
           sock.write(p.bb);//发送到服务端
           if (!p.bb.hasRemaining()) {
              sentCount++;
              outgoingQueue.removeFirstOccurrence(p);
              if (p.requestHeader != null
                    && p.requestHeader.getType() != OpCode.ping
                    && p.requestHeader.getType() != OpCode.auth) {
                 synchronized (pendingQueue) {
                    pendingQueue.add(p);
                 }
              }
           }
        }
        if (outgoingQueue.isEmpty()) {
           disableWrite();//如果没有要发的就禁止写
        } else if (!initialized && p != null && !p.bb.hasRemaining()) {
           disableWrite();.//禁止写
        } else {
           // Just in case
           enableWrite();//开启写
        }
     }
  }
}
复制代码