单机版Zookeeper新建连接交互流程源码分析

2023-12-26 09:09:29

Zookeeper连接交互流程概述

Zookeeper是一个分布式协调服务，它为分布式应用程序提供协调服务，包括数据同步、命名服务、配置管理等。Zookeeper使用一种称为Zab协议的共识算法来保证数据的一致性。

Zookeeper的连接交互流程可以分为以下几个步骤：

Client端向Server端发送连接请求。
Server端接收连接请求并进行处理。
Server端向Client端发送连接响应。
Client端接收Server端的连接响应并建立连接。

单机版Zookeeper新建连接交互流程源码分析

Server接收处理及响应

public void handleConnectRequest(ServerCnxn cnxn) throws IOException {
    if (!allowSystemConnections) {
        // don't allow connects from outside the ACL
        cnxn.close();
        return;
    }
    synchronized (cnxns) {
        if (cnxns.size() >= maxClientCnxns) {
            // too many connections
            close(cnxn);
            return;
        }
        cnxns.add(cnxn);
        // don't register ephemeral nodes until after the connection is closed
        if (!closeWait) {
            cnxn.EphemeralConnections.process(cnxns);
        }
        // ZOOKEEPER-706
        // give a 10 minute grace period before closing ephemeral connections
        // on this server
        cnxn.closeWait = closeWait;
    }
    LOG.info("Connected client " + cnxn.sessionid + " from " + cnxn.remoteAddr());
    Thread.yield();
    // Do this rather than deferring until after the constructor has
    // returned.  That's because sync's on the Connection object may
    // occur before construction is complete, e.g. in cases where a
    // Session constructor calls into ZooKeeper methods.
    cnxn.sendResponse(response);
}

在handleConnectRequest方法中，Server端首先检查连接请求是否来自允许连接的ACL列表，如果不在允许连接的ACL列表中，则关闭连接。

然后，Server端检查当前连接数是否超过最大连接数，如果超过最大连接数，则关闭连接。

如果当前连接数没有超过最大连接数，则将连接添加到连接列表中，并向Client端发送连接响应。

Client端接收Server端响应

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException, InterruptedException {
    ...
    long sessionId = joinServer(connectString, sessionTimeout);
    ...
}

private long joinServer(String connectString, int sessionTimeout) throws IOException, InterruptedException {
    ...
    sendConnectRequest(connectString, sessionTimeout);
    ...
}

private void sendConnectRequest(String connectString, int sessionTimeout) throws IOException, InterruptedException {
    ...
    String[] hosts = hostPortList.split(",");
    for (String hp : hosts) {
        ServerCnxn cnxn = new ServerCnxn(hp, port, sessionId, Watcher.NONE, sessionTimeout, connectTimeout, readTimeout, clientCnxnSocket, 0);
        connectOne(cnxn);
        // if the client is connected, set the session id and return
        if (cnxn.getSessionId() != 0) {
            sessions.put(cnxn.getSessionId(), cnxn);
            connected.set(true);
            return cnxn.getSessionId();
        }
        closeCnxn(cnxn);
    }
    ...
}