public InetSocketAddress next(long spinDelay) {
    ++currentIndex;
    if (currentIndex == serverAddresses.size()) {
        currentIndex = 0;
    }
    if (currentIndex == lastIndex && spinDelay > 0) {
        try {
            Thread.sleep(spinDelay);
        } catch (InterruptedException e) {
            LOG.warn("Unexpected exception", e);
        }
    } else if (lastIndex == -1) {
        // We don't want to sleep on the first ever connect attempt.
        lastIndex = 0;
    }

    return serverAddresses.get(currentIndex);
}

public void create(final String path, byte data[], List<ACL> acl,
            CreateMode createMode,  StringCallback cb, Object ctx)
{
    final String clientPath = path;
    PathUtils.validatePath(clientPath, createMode.isSequential());

    final String serverPath = prependChroot(clientPath);

    RequestHeader h = new RequestHeader();
    // 设置请求类型，zk server端会根据这个类型得知客户端向进行何种操作
    h.setType(ZooDefs.OpCode.create);
    CreateRequest request = new CreateRequest();
    CreateResponse response = new CreateResponse();
    ReplyHeader r = new ReplyHeader();
    // 组装请求对象，在这里可以看出zk并没有将StringCallback的实现类组织到request对象中
    request.setData(data);
    request.setFlags(createMode.toFlag());
    request.setPath(serverPath);
    request.setAcl(acl);
    // 将请求压入sendingQueue队列中
    cnxn.queuePacket(h, r, request, response, cb, clientPath,
            serverPath, ctx, null);
}

Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
            Record response, AsyncCallback cb, String clientPath,
            String serverPath, Object ctx, WatchRegistration watchRegistration)
{
    Packet packet = null;

    // Note that we do not generate the Xid for the packet yet. It is
    // generated later at send-time, by an implementation of ClientCnxnSocket::doIO(),
    // where the packet is actually sent.
    synchronized (outgoingQueue) {
		// 从这里可以看出request对象和cb仍然没有任何关联
        packet = new Packet(h, r, request, response, watchRegistration);
        packet.cb = cb;
        packet.ctx = ctx;
        packet.clientPath = clientPath;
        packet.serverPath = serverPath;
        if (!state.isAlive() || closing) {
            conLossPacket(packet);
        } else {
            // If the client is asking to close the session then
            // mark as closing
            if (h.getType() == OpCode.closeSession) {
                closing = true;
            }
			// packet对象入队
            outgoingQueue.add(packet);
        }
    }
    sendThread.getClientCnxnSocket().wakeupCnxn();
    return packet;
}

synchronized(outgoingQueue) {
    Packet p = findSendablePacket(outgoingQueue,
            cnxn.sendThread.clientTunneledAuthenticationInProgress());

    if (p != null) {
        updateLastSend();
        // If we already started writing p, p.bb will already exist
        if (p.bb == null) {
            if ((p.requestHeader != null) &&
                    (p.requestHeader.getType() != OpCode.ping) &&
                    (p.requestHeader.getType() != OpCode.auth)) {
                p.requestHeader.setXid(cnxn.getXid());
            }
            p.createBB();
        }
		// 向socket连接写入数据
        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) {
                    // 将packet对象入队
                    pendingQueue.add(p);
                }
            }
        }
    }
    if (outgoingQueue.isEmpty()) {
        // No more packets to send: turn off write interest flag.
        // Will be turned on later by a later call to enableWrite(),
        // from within ZooKeeperSaslClient (if client is configured
        // to attempt SASL authentication), or in either doIO() or
        // in doTransport() if not.
        disableWrite();
    } else if (!initialized && p != null && !p.bb.hasRemaining()) {
        // On initial connection, write the complete connect request
        // packet, but then disable further writes until after
        // receiving a successful connection response.  If the
        // session is expired, then the server sends the expiration
        // response and immediately closes its end of the socket.  If
        // the client is simultaneously writing on its end, then the
        // TCP stack may choose to abort with RST, in which case the
        // client would never receive the session expired event.  See
        // http://docs.oracle.com/javase/6/docs/technotes/guides/net/articles/connection_release.html
        disableWrite();
    } else {
        // Just in case
        enableWrite();
    }
}

public void createBB() {
    try {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
        boa.writeInt(-1, "len"); // We'll fill this in later
        if (requestHeader != null) {
            requestHeader.serialize(boa, "header");
        }
        if (request instanceof ConnectRequest) {
            request.serialize(boa, "connect");
            // append "am-I-allowed-to-be-readonly" flag
            boa.writeBool(readOnly, "readOnly");
        } else if (request != null) {
            request.serialize(boa, "request");
        }
        baos.close();
        this.bb = ByteBuffer.wrap(baos.toByteArray());
        this.bb.putInt(this.bb.capacity() - 4);
        this.bb.rewind();
    } catch (IOException e) {
        LOG.warn("Ignoring unexpected exception", e);
    }
}

if (cmd.equals("ls") && args.length >= 2) {
    path = args[1];
    List<String> children = zk.getChildren(path, watch);
    System.out.println(children);
}