ZooKeeper客户端源码（一）——向服务端建立连接+会话建立+心跳保持长连接

String connectString = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";
ZooKeeper zooKeeper = new ZooKeeper(connectString, 20000, new Watcher() {
    @Override
    public void process(WatchedEvent event) {
        if (event.getState() == Event.KeeperState.SyncConnected) {
            System.out.println("会话建立成功");
        }
    }
});

// org.apache.zookeeper.client.StaticHostProvider#shuffle
private List<InetSocketAddress> shuffle(Collection<InetSocketAddress> serverAddresses) {
    List<InetSocketAddress> tmpList = new ArrayList<>(serverAddresses.size());
    tmpList.addAll(serverAddresses);
    Collections.shuffle(tmpList, sourceOfRandomness);
    return tmpList;
}

Random sourceOfRandomness = new Random(System.currentTimeMillis() ^ this.hashCode())

public InetSocketAddress next(long spinDelay) {
    boolean needToSleep = false;
    InetSocketAddress addr;

    synchronized (this) {
        // 省略部分无关代码
        
        // currentIndex自增，如果等于服务地址列表长度，就重置为0
        ++currentIndex;
        if (currentIndex == serverAddresses.size()) {
            currentIndex = 0;
        }
        addr = serverAddresses.get(currentIndex);
        // 两个游标currentIndex、lastIndex
        // currentIndex 当前选择的位置，lastIndex上次选择的位置
        // lastIndex 什么时候设置呢？会话建立成功之后调用 onConnected，将currentIndex赋值给lastIndex
        needToSleep = needToSleep || (currentIndex == lastIndex && spinDelay > 0);
        if (lastIndex == -1) {
            lastIndex = 0;
        }
    }
    // 如果 currentIndex和lastIndex且spinDelay>0，就需要休眠spinDelay时间，
    if (needToSleep) {
        try {
            Thread.sleep(spinDelay);
        } catch (InterruptedException e) {
            LOG.warn("Unexpected exception", e);
        }
    }
    // 解析InetSocketAddress，
    // 如果一个主机映射了多个ip地址（InetAddress）
    // 就打乱选择其中一个地址返回
    return resolve(addr);
}

public ConnectStringParser(String connectString) {
    // parse out chroot, if any
    // 解析chroot
    // connectString 可以指定某个路径，
    // 如127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183/test
    int off = connectString.indexOf('/');
    if (off >= 0) {
        String chrootPath = connectString.substring(off);
        // ignore "/" chroot spec, same as null
        if (chrootPath.length() == 1) {
            this.chrootPath = null;
        } else {
            PathUtils.validatePath(chrootPath);
            this.chrootPath = chrootPath;
        }
        connectString = connectString.substring(0, off);
    } else {
        this.chrootPath = null;
    }
    // 按 , 分割
    List<String> hostsList = split(connectString, ",");
    for (String host : hostsList) {
        int port = DEFAULT_PORT;
        String[] hostAndPort = NetUtils.getIPV6HostAndPort(host);
        if (hostAndPort.length != 0) {
            host = hostAndPort[0];
            if (hostAndPort.length == 2) {
                port = Integer.parseInt(hostAndPort[1]);
            }
        } else {
            int pidx = host.lastIndexOf(':');
            if (pidx >= 0) {
                // otherwise : is at the end of the string, ignore
                if (pidx < host.length() - 1) {
                    port = Integer.parseInt(host.substring(pidx + 1));
                }
                host = host.substring(0, pidx);
            }
        }
        // 封装未解析的InetSocketAddress
        serverAddresses.add(InetSocketAddress.createUnresolved(host, port));
    }
}

ConnectRequest conReq = new ConnectRequest(0, lastZxid, sessionTimeout, sessId, sessionPasswd);

Packet(RequestHeader requestHeader,ReplyHeader replyHeader,Record request,
    Record response,
    WatchRegistration watchRegistration,
    boolean readOnly)

outgoingQueue.addFirst(new Packet(null, null, conReq, null, null, readOnly));

// clientCnxnSocket.getIdleSend为距离上次发送的时长
// readTimeout = sessionTimeout * 2 / 3
int timeToNextPing = readTimeout / 2
                     - clientCnxnSocket.getIdleSend()
                     - ((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);