ZooKeeper 的网络通信协议详解

ZooKeeper 协议简述

ZooKeeper 协议的底层实现

客户端请求头底层解析

class RequestHeader implements Record{

  private int xid;

  private int type;

}

会话创建

ConnectRequest implements Record {

  private int protocolVersion;

  private long lastZxidSeen;

  private int timeOut;

  private long sessionId;

  private byte[] passwd;

 }

节点查询

public class GetDataRequest implements Record {

  private String path;

  private boolean watch;

节点更新

public class SetDataRequest implements Record {

  private String path;

  private byte[] data;

  private int version;

 }

响应协议

public class ReplyHeader implements Record {

  private int xid;

  private long zxid;

  private int err;

  }

服务端请求体解析

响应会话创建

public class ConnectResponse implements Record {

  private int protocolVersion;

  private int timeOut;

  private long sessionId;

  private byte[] passwd;

  }

public class GetDataResponse implements Record {

  private byte[] data;

  private org.apache.zookeeper.data.Stat stat;

  }

响应节点更新

public class SetDataResponse implements Record {

  private org.apache.zookeeper.data.Stat stat;

  }

会话属性

• 会话 ID：会话 ID 作为一个会话的标识符，当我们创建一次会话的时候，ZooKeeper 会自动为其分配一个唯一的 ID 编码。

• 会话超时时间：
一般来说，一个会话的超时时间就是指一次会话从发起后到被服务器关闭的时长。
而设置会话超时时间后，服务器会参考设置的超时时间，最终计算一个服务端自己的超时时间。
zookeeper会话超时设置的坑
如果一个会话的超时时间是在服务器端实现的。会话超时服务器端会主动关闭会话。
所以在会话请求的请求体中包括了设置会话超时的属性字段，而我们经常会遇到一个问题就是明明通过客户端将超时时间设置了一个值，而在实际执行的时候会话超时的时间可能远远小于我们设置的时间。
这是因为在 ZooKeeper 处理会话超时时间的时候不只是简单地使用客户端传来的超时时间还会根据 minSessionTimeout 和 maxSessionTimeout 这两个参数来调整时间的设置
而这个超时时间则是最终真正用于 ZooKeeper 中服务端用户会话管理的超时时间。

• 会话关闭状态：会话关闭 isClosing 状态属性字段表示一个会话是否已经关闭。
如果服务器检查到一个会话已经因为超时等原因失效时， ZooKeeper 会在该会话的 isClosing 属性值标记为关闭，再之后就不对该会话进行操作了。

会话状态

会话底层实现

class SessionTrackerImpl implements SessionTracker{

 ConcurrentHashMap<Long, SessionImpl> sessionsById；

 ConcurrentMap<Long, Integer> sessionsWithTimeout;

}

• sessionsById，用于根据会话 ID 来管理具体的会话实体。

• sessionsWithTimeout，根据不同的会话 ID 管理每个会话的超时时间。

public static long initializeNextSession(long id) {

    long nextSid;

    nextSid = (Time.currentElapsedTime() << 24) >>> 8;

    nextSid =  nextSid | (id <<56);

    return nextSid;

 }

会话异常

会话管理策略

底层实现

public class ExpiryQueue<E> {

    private final ConcurrentHashMap<E, Long> elemMap;

    public Long remove(E elem) {...}

    public Long update(E elem, int timeout) {...}

    public Set<E> poll() {...}

}

private long roundToNextInterval(long time) {

        return (time / expirationInterval + 1) * expirationInterval;

}

public void run() {

    try {

        while (running) {

            long waitTime = sessionExpiryQueue.getWaitTime();

            if (waitTime > 0) {

                Thread.sleep(waitTime);

                continue;

            }

            for (SessionImpl s : sessionExpiryQueue.poll()) {

                setSessionClosing(s.sessionId);

                expirer.expire(s);

            }

        }

 ...

}

private void close(long sessionId) {

  Request si = new Request(null,sessionId,0,OpCode.closeSession, null, null);

    setLocalSessionFlag(si);

    submitRequest(si);

    }

服务端是如何处理一次会话请求的？

流程

底层实现

请求预处理器

public class PrepRequestProcessor extends Thread implements RequestProcessor {

  RequestProcessor nextProcessor;

}

protected void pRequest(Request request) throws RequestProcessorException {

 switch (request.type) {

    case OpCode.create:

     CreateRequest createRequest = new CreateRequest();

     pRequest2Txn(request.type, zks.getNextZxid(), request,createRequest, true);

     break;

    case OpCode.delete:

 }

}

事物处理器

Sync 流程

Proposal 流程

Commit 流程

class CommitProcessor {

  LinkedList queuedRequests

}

synchronized public void commit(Request request) {

  committedRequests.add(request);

  notifyAll();

}

最终处理器