ZooKeeper 单机服务端与客户端连接交互流程分析

后端

2024-01-19 07:01:11

ZooKeeper 连接交互流程

ZooKeeper 是一个分布式协调服务，它为分布式应用提供统一的命名服务、配置管理、集群管理等功能。ZooKeeper 采用的是主从复制架构，由一个主节点和多个从节点组成。主节点负责处理客户端的请求，而从节点则负责复制主节点的数据。

1. 客户端发起连接

当客户端需要使用 ZooKeeper 服务时，它首先会向 ZooKeeper 服务端发起连接请求。连接请求中包含客户端的 IP 地址、端口号等信息。

2. 服务端响应

ZooKeeper 服务端收到客户端的连接请求后，会进行验证和授权。如果验证和授权通过，服务端会向客户端发送一个连接响应。连接响应中包含服务端的 IP 地址、端口号等信息。

3. 客户端接收响应

客户端收到服务端的连接响应后，会解析响应中的信息，并建立与服务端的连接。

源码分析

1. 服务端响应

public void processConnectRequest(Request request) {
    ZooKeeperServer zkServer = server;
    if (!zkServer.authManager.authenticate(request.authInfo,
            "digest",
            request.sessionId)) {
        return;
    }
    long sessionId = request.sessionId;
    // If it is a session reconnect request (new client connecting with a
    // session id)
    if (request.sessionId != 0) {
        // Verify if this is a valid session
        Session session = zkServer.sessionTracker.getSession(sessionId);
        if (session != null) {
            session.updateJMXNode();
        } else {
            zkServer.sessionTracker.killSession(sessionId);
        }
    }
    try {
        if (request.sessionId == 0) {
            long id = zkServer.sessionTracker.createSession(zkServer);
            outBuffer.writeSessionMessage(id, request.timeOut);
        } else {
            // session reconnect
            zkServer.sessionTracker.touchSession(sessionId, zkServer);
            outBuffer.writeSessionMessage(0, request.timeOut);
        }
    } catch (KeeperException e) {
        outBuffer.writeExceptionMessage(e);
    }
}

在上面的代码中，processConnectRequest() 方法是处理客户端连接请求的入口函数。它首先验证客户端的授权信息，如果授权通过，则继续处理。然后，它检查客户端的连接类型，如果是新客户端连接，则创建一个新的会话，否则更新现有会话的超时时间。最后，它向客户端发送连接响应消息。

2. 客户端接收响应

protected void processConnectResponse(ByteBuffer request) {
    if (request.getInt() != 0) {
        // the server will not set the session Id to 0
        throw new IOException("Session establish error, HResultCode: " + request.getInt());
    }
    long sessionId = request.getLong();
    this.sessionId = sessionId;
}