private void readConnectRequest() throws IOException, InterruptedException {
    if (!isZKServerRunning()) {
        throw new IOException("ZooKeeperServer not running");
    }
    zkServer.processConnectRequest(this, incomingBuffer);
    initialized = true;
}

public void processConnectRequest(ServerCnxn cnxn, ByteBuffer incomingBuffer)
  throws IOException {
  // 读取数据相关对象构造
  BinaryInputArchive bia =
    BinaryInputArchive.getArchive(new ByteBufferInputStream(incomingBuffer));
  ConnectRequest connReq = new ConnectRequest();
  // 反序列化 connect 数据
  connReq.deserialize(bia, "connect");
  if (LOG.isDebugEnabled()) {
    LOG.debug("Session establishment request from client "
              + cnxn.getRemoteSocketAddress()
              + " client's lastZxid is 0x"
              + Long.toHexString(connReq.getLastZxidSeen()));
  }
  // 判断是否只读
  boolean readOnly = false;
  try {
    readOnly = bia.readBool("readOnly");
    cnxn.isOldClient = false;
  } catch (IOException e) {
    LOG.warn("Connection request from old client "
             + cnxn.getRemoteSocketAddress()
             + "; will be dropped if server is in r-o mode");
  }
  // 非只读并且当前zk服务是只读服务抛出异常
  if (!readOnly && this instanceof ReadOnlyZooKeeperServer) {
    String msg = "Refusing session request for not-read-only client "
      + cnxn.getRemoteSocketAddress();
    LOG.info(msg);
    throw new CloseRequestException(msg);
  }
  // 判断连接器请求中的zxid是否大于数据树中的最大zxid，如果是则抛出异常
  if (connReq.getLastZxidSeen() > zkDb.dataTree.lastProcessedZxid) {
    String msg = "Refusing session request for client "
      + cnxn.getRemoteSocketAddress()
      + " as it has seen zxid 0x"
      + Long.toHexString(connReq.getLastZxidSeen())
      + " our last zxid is 0x"
      + Long.toHexString(getZKDatabase().getDataTreeLastProcessedZxid())
      + " client must try another server";

    LOG.info(msg);
    throw new CloseRequestException(msg);
  }
  // 读取会话超时时间
  int sessionTimeout = connReq.getTimeOut();
  // 读取密码
  byte passwd[] = connReq.getPasswd();
  // 超时时间相关处理
  int minSessionTimeout = getMinSessionTimeout();
  if (sessionTimeout < minSessionTimeout) {
    sessionTimeout = minSessionTimeout;
  }
  int maxSessionTimeout = getMaxSessionTimeout();
  if (sessionTimeout > maxSessionTimeout) {
    sessionTimeout = maxSessionTimeout;
  }
  // 设置超时时间
  cnxn.setSessionTimeout(sessionTimeout);
  // 在会话建立完成之前禁止接收数据
  cnxn.disableRecv();
  // 获取session id
  long sessionId = connReq.getSessionId();
  // 如果session id 等于0
  if (sessionId == 0) {
    // 重算session id
    long id = createSession(cnxn, passwd, sessionTimeout);
    LOG.debug("Client attempting to establish new session:" +
              " session = 0x{}, zxid = 0x{}, timeout = {}, address = {}",
              Long.toHexString(id),
              Long.toHexString(connReq.getLastZxidSeen()),
              connReq.getTimeOut(),
              cnxn.getRemoteSocketAddress());
  }
  // session id 非零的情况
  else {
    // 从连接请求中获取session id
    long clientSessionId = connReq.getSessionId();
    LOG.debug("Client attempting to renew session:" +
              " session = 0x{}, zxid = 0x{}, timeout = {}, address = {}",
              Long.toHexString(clientSessionId),
              Long.toHexString(connReq.getLastZxidSeen()),
              connReq.getTimeOut(),
              cnxn.getRemoteSocketAddress());
    // 如果serverCnxnFactory不为空关闭session
    if (serverCnxnFactory != null) {
      serverCnxnFactory.closeSession(sessionId);
    }
    // 如果secureServerCnxnFactory不为空关闭session
    if (secureServerCnxnFactory != null) {
      secureServerCnxnFactory.closeSession(sessionId);
    }
    // 设置session id
    cnxn.setSessionId(sessionId);
    // 重新打开session
    reopenSession(cnxn, sessionId, passwd, sessionTimeout);
  }
}

public void reopenSession(ServerCnxn cnxn, long sessionId, byte[] passwd,
                          int sessionTimeout) throws IOException {
  // 检查密码
  if (checkPasswd(sessionId, passwd)) {
    // 验证会话
    revalidateSession(cnxn, sessionId, sessionTimeout);
  }
  else {
    LOG.warn("Incorrect password from " + cnxn.getRemoteSocketAddress()
             + " for session 0x" + Long.toHexString(sessionId));
    // 完成会话创建
    finishSessionInit(cnxn, false);
  }
}

public void finishSessionInit(ServerCnxn cnxn, boolean valid) {
  try {
    // 如果验证通过
    if (valid) {
      // 注册连接对象
      if (serverCnxnFactory != null && serverCnxnFactory.cnxns.contains(cnxn)) {
        serverCnxnFactory.registerConnection(cnxn);
      } else if (secureServerCnxnFactory != null
                 && secureServerCnxnFactory.cnxns.contains(cnxn)) {
        secureServerCnxnFactory.registerConnection(cnxn);
      }
    }
  } catch (Exception e) {
    LOG.warn("Failed to register with JMX", e);
  }

  try {
    // 创建连接响应对象
    ConnectResponse rsp = new ConnectResponse(0, valid ? cnxn.getSessionTimeout()
                                              : 0, valid ? cnxn.getSessionId() : 0, // send 0 if session is no
                                              // longer valid
                                              valid ? generatePasswd(cnxn.getSessionId()) : new byte[16]);
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
    bos.writeInt(-1, "len");
    // 连接对象写入到bos
    rsp.serialize(bos, "connect");
    // 判断是否是老版本的客户端
    if (!cnxn.isOldClient) {
      // 设置是否只读
      bos.writeBool(
        this instanceof ReadOnlyZooKeeperServer, "readOnly");
    }
    // 关闭 baos
    baos.close();
    // 将baos转换为字节流通过连接对象发出去
    ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
    bb.putInt(bb.remaining() - 4).rewind();
    cnxn.sendBuffer(bb);

    // 验证通过
    if (valid) {
      LOG.debug("Established session 0x"
                + Long.toHexString(cnxn.getSessionId())
                + " with negotiated timeout " + cnxn.getSessionTimeout()
                + " for client "
                + cnxn.getRemoteSocketAddress());
      // 恢复数据接收
      cnxn.enableRecv();
    } else {

      LOG.info("Invalid session 0x"
               + Long.toHexString(cnxn.getSessionId())
               + " for client "
               + cnxn.getRemoteSocketAddress()
               + ", probably expired");
      // 发送关闭连接信息
      cnxn.sendBuffer(ServerCnxnFactory.closeConn);
    }

  } catch (Exception e) {
    LOG.warn("Exception while establishing session, closing", e);
    // 关闭连接对象
    cnxn.close();
  }
}

interface Session {
  long getSessionId();

  int getTimeout();

  boolean isClosing();
}

interface SessionExpirer {
  void expire(Session session);

  long getServerId();
}

public SessionTrackerImpl(SessionExpirer expirer,
                          ConcurrentMap<Long, Integer> sessionsWithTimeout, int tickTime,
                          long serverId, ZooKeeperServerListener listener) {
    super("SessionTracker", listener);
    this.expirer = expirer;
    this.sessionExpiryQueue = new ExpiryQueue<SessionImpl>(tickTime);
    this.sessionsWithTimeout = sessionsWithTimeout;
    this.nextSessionId.set(initializeNextSession(serverId));
    for (Entry<Long, Integer> e : sessionsWithTimeout.entrySet()) {
        addSession(e.getKey(), e.getValue());
    }

    EphemeralType.validateServerId(serverId);
}

public static long initializeNextSession(long id) {

  long nextSid;
  // 当前时间(毫秒)左移24位，无符号右移8位
  nextSid = (Time.currentElapsedTime() << 24) >>> 8;
  // sid 与id左移56位后或运算
  nextSid = nextSid | (id << 56);
  // 判断计算的sid是否和最小long值相同，如果相同则进行加一操作
  if (nextSid == EphemeralType.CONTAINER_EPHEMERAL_OWNER) {
    ++nextSid; 
  }
  return nextSid;
}

public synchronized boolean addSession(long id, int sessionTimeout) {
  // 向sessionsWithTimeout容器中加入数据
  sessionsWithTimeout.put(id, sessionTimeout);

  boolean added = false;

  // 从sessionsById容器中根据session id 获取session
  SessionImpl session = sessionsById.get(id);
  // 不存在创建
  if (session == null) {
    session = new SessionImpl(id, sessionTimeout);
  }

  // 向sessionsById容器中加入session
  SessionImpl existedSession = sessionsById.putIfAbsent(id, session);

  if (existedSession != null) {
    session = existedSession;
  }
  else {
    added = true;
    LOG.debug("Adding session 0x" + Long.toHexString(id));
  }

  if (LOG.isTraceEnabled()) {
    String actionStr = added ? "Adding" : "Existing";
    ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
                             "SessionTrackerImpl --- " + actionStr + " session 0x"
                             + Long.toHexString(id) + " " + sessionTimeout);
  }

  // 更新session过期信息
  updateSessionExpiry(session, sessionTimeout);
  return added;
}

private void updateSessionExpiry(SessionImpl s, int timeout) {
  logTraceTouchSession(s.sessionId, timeout, "");
  sessionExpiryQueue.update(s, timeout);
}

synchronized public Map<Long, Set<Long>> getSessionExpiryMap() {
    Map<Long, Set<SessionImpl>> expiryMap = sessionExpiryQueue.getExpiryMap();
    Map<Long, Set<Long>> sessionExpiryMap = new TreeMap<Long, Set<Long>>();
    for (Entry<Long, Set<SessionImpl>> e : expiryMap.entrySet()) {
        Set<Long> ids = new HashSet<Long>();
        sessionExpiryMap.put(e.getKey(), ids);
        for (SessionImpl s : e.getValue()) {
            ids.add(s.sessionId);
        }
    }
    return sessionExpiryMap;
}

synchronized public boolean touchSession(long sessionId, int timeout) {
  // 根据session id 获取session
  SessionImpl s = sessionsById.get(sessionId);

  // session 为空返回false
  if (s == null) {
    logTraceTouchInvalidSession(sessionId, timeout);
    return false;
  }

  // session 处于关闭状态 返回false
  if (s.isClosing()) {
    logTraceTouchClosingSession(sessionId, timeout);
    return false;
  }

  // 更新过期时间
  updateSessionExpiry(s, timeout);
  return true;
}

private void updateSessionExpiry(SessionImpl s, int timeout) {
  logTraceTouchSession(s.sessionId, timeout, "");
  sessionExpiryQueue.update(s, timeout);
}

synchronized public void setSessionClosing(long sessionId) {
    if (LOG.isTraceEnabled()) {
        LOG.trace("Session closing: 0x" + Long.toHexString(sessionId));
    }
    SessionImpl s = sessionsById.get(sessionId);
    if (s == null) {
        return;
    }
    s.isClosing = true;
}

synchronized public void removeSession(long sessionId) {
    LOG.debug("Removing session 0x" + Long.toHexString(sessionId));
    SessionImpl s = sessionsById.remove(sessionId);
    sessionsWithTimeout.remove(sessionId);
    if (LOG.isTraceEnabled()) {
        ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
                "SessionTrackerImpl --- Removing session 0x"
                        + Long.toHexString(sessionId));
    }
    if (s != null) {
        sessionExpiryQueue.remove(s);
    }
}

public synchronized void checkSession(long sessionId, Object owner)
        throws KeeperException.SessionExpiredException,
        KeeperException.SessionMovedException,
        KeeperException.UnknownSessionException {
    LOG.debug("Checking session 0x" + Long.toHexString(sessionId));
    // 获取 session
    SessionImpl session = sessionsById.get(sessionId);

    // session 为空抛出异常
    if (session == null) {
        throw new KeeperException.UnknownSessionException();
    }

    // session 已关闭抛出异常
    if (session.isClosing()) {
        throw new KeeperException.SessionExpiredException();
    }

    // 所有者为空赋值所有者
    if (session.owner == null) {
        session.owner = owner;
    } 
    // session 所有者和参数所有者不匹配抛出异常
    else if (session.owner != owner) {
        throw new KeeperException.SessionMovedException();
    }
}

@Override
public void run() {
  try {
    while (running) {
      // 获取等待时间
      long waitTime = sessionExpiryQueue.getWaitTime();
      if (waitTime > 0) {
        Thread.sleep(waitTime);
        continue;
      }

      // 从session过期队列中获取session
      for (SessionImpl s : sessionExpiryQueue.poll()) {
        // 设置关闭状态
        setSessionClosing(s.sessionId);
        // 进行过期接口操作
        expirer.expire(s);
      }
    }
  } catch (InterruptedException e) {
    handleException(this.getName(), e);
  }
  LOG.info("SessionTrackerImpl exited loop!");
}

public long getWaitTime() {
  long now = Time.currentElapsedTime();
  long expirationTime = nextExpirationTime.get();
  return now < expirationTime ? (expirationTime - now) : 0L;
}

public void expire(Session session) {
  long sessionId = session.getSessionId();
  LOG.info("Expiring session 0x" + Long.toHexString(sessionId) + ", timeout of "
           + session.getTimeout() + "ms exceeded");
  close(sessionId);
}

public ExpiryQueue(int expirationInterval) {
    this.expirationInterval = expirationInterval;
    nextExpirationTime.set(roundToNextInterval(Time.currentElapsedTime()));
}

private long roundToNextInterval(long time) {
    return (time / expirationInterval + 1) * expirationInterval;
}

public Long update(E elem, int timeout) {
  // 从elemMap容器中根据元素获取过期时间，含义为上一个过期时间
  Long prevExpiryTime = elemMap.get(elem);
  // 获取当前时间
  long now = Time.currentElapsedTime();
  // 计算下一个过期时间
  Long newExpiryTime = roundToNextInterval(now + timeout);

  // 判断新的过期时间和上一个过期时间是否相同，如果相同则返回null
  if (newExpiryTime.equals(prevExpiryTime)) {
    return null;
  }

  // First add the elem to the new expiry time bucket in expiryMap.
  // 获取过期时间对应的数据
  Set<E> set = expiryMap.get(newExpiryTime);
  if (set == null) {
    // 创建set对象
    set = Collections.newSetFromMap(
      new ConcurrentHashMap<E, Boolean>());
    // 向expiryMap容器中加入过期数据
    Set<E> existingSet = expiryMap.putIfAbsent(newExpiryTime, set);
    // 存在历史过期数据将其赋值到set变量
    if (existingSet != null) {
      set = existingSet;
    }
  }
  // 将参数数据加入到set集合中
  set.add(elem);

  // 向elemMap容器加入elem数据，并获取历史过期时间
  prevExpiryTime = elemMap.put(elem, newExpiryTime);
  // 历史过期时间不为空
  // 新的过期时间和历史过期时间不相同
  if (prevExpiryTime != null && !newExpiryTime.equals(prevExpiryTime)) {
    // 根据历史过期时间获取过期元素集合
    Set<E> prevSet = expiryMap.get(prevExpiryTime);
    // 过期元素集合不为空的情况下移除当前元素
    if (prevSet != null) {
      prevSet.remove(elem);
    }
  }
  return newExpiryTime;
}

public Set<E> poll() {
  // 获取当前时间
  long now = Time.currentElapsedTime();
  // 获取过期时间
  long expirationTime = nextExpirationTime.get();
  // 如果当前时间小于过期时间，返回空集合
  if (now < expirationTime) {
    return Collections.emptySet();
  }

  // 创建存储结果的容器
  Set<E> set = null;
  // 计算新的过期时间
  long newExpirationTime = expirationTime + expirationInterval;
  // 将新的过期时间赋值到变量nextExpirationTime中
  if (nextExpirationTime.compareAndSet(
    expirationTime, newExpirationTime)) {
    // 从expiryMap容器中移除过期时间对应的数据，并将数据保存到结果容器中
    set = expiryMap.remove(expirationTime);
  }
  // 如果结果存储容器为空创建空集合
  if (set == null) {
    return Collections.emptySet();
  }
  return set;
}

public class LocalSessionTracker extends SessionTrackerImpl {
  public LocalSessionTracker(SessionExpirer expirer,
                             ConcurrentMap<Long, Integer> sessionsWithTimeouts,
                             int tickTime, long id, ZooKeeperServerListener listener) {
    super(expirer, sessionsWithTimeouts, tickTime, id, listener);
  }

  /**
    * 检查session id 是否位于sessionsById容器中
    */
  public boolean isLocalSession(long sessionId) {
    return isTrackingSession(sessionId);
  }

  public boolean isGlobalSession(long sessionId) {
    return false;
  }

  /**
    * 恒定抛出异常
    */
  public boolean addGlobalSession(long sessionId, int sessionTimeout) {
    throw new UnsupportedOperationException();
  }
}

public boolean isTrackingSession(long sessionId) {
    return isLocalSession(sessionId) || isGlobalSession(sessionId);
}

public boolean isLocalSession(long sessionId) {
    return localSessionTracker != null &&
            localSessionTracker.isTrackingSession(sessionId);
}

public int upgradeSession(long sessionId) {
  // 容器localSessionsWithTimeouts为空返回-1
  if (localSessionsWithTimeouts == null) {
    return -1;
  }
  // 从localSessionsWithTimeouts容器中根据session id移除数据，暂存超时时间
  Integer timeout = localSessionsWithTimeouts.remove(sessionId);
  // 超时时间不为空
  if (timeout != null) {
    LOG.info("Upgrading session 0x" + Long.toHexString(sessionId));
    // 将session 放入到全局容器中
    addGlobalSession(sessionId, timeout);
    // 移除本地session
    localSessionTracker.removeSession(sessionId);
    // 返回超时时间
    return timeout;
  }
  return -1;
}

public LearnerSessionTracker(SessionExpirer expirer,
                             ConcurrentMap<Long, Integer> sessionsWithTimeouts,
                             int tickTime, long id, boolean localSessionsEnabled,
                             ZooKeeperServerListener listener) {
  this.expirer = expirer;
  this.touchTable.set(new ConcurrentHashMap<Long, Integer>());
  this.globalSessionsWithTimeouts = sessionsWithTimeouts;
  this.serverId = id;
  nextSessionId.set(SessionTrackerImpl.initializeNextSession(serverId));

  this.localSessionsEnabled = localSessionsEnabled;
  if (this.localSessionsEnabled) {
    createLocalSessionTracker(expirer, tickTime, id, listener);
  }
}

public boolean isGlobalSession(long sessionId) {
    return globalSessionsWithTimeouts.containsKey(sessionId);
}

public boolean addGlobalSession(long sessionId, int sessionTimeout) {
  boolean added =
    globalSessionsWithTimeouts.put(sessionId, sessionTimeout) == null;
  if (localSessionsEnabled && added) {
    LOG.info("Adding global session 0x" + Long.toHexString(sessionId));
  }
  touchTable.get().put(sessionId, sessionTimeout);
  return added;
}

public LeaderSessionTracker(SessionExpirer expirer,
                            ConcurrentMap<Long, Integer> sessionsWithTimeouts,
                            int tickTime, long id, boolean localSessionsEnabled,
                            ZooKeeperServerListener listener) {

  this.globalSessionTracker = new SessionTrackerImpl(
    expirer, sessionsWithTimeouts, tickTime, id, listener);

  this.localSessionsEnabled = localSessionsEnabled;
  if (this.localSessionsEnabled) {
    createLocalSessionTracker(expirer, tickTime, id, listener);
  }
  serverId = id;
}

public boolean isGlobalSession(long sessionId) {
  return globalSessionTracker.isTrackingSession(sessionId);
}

public boolean addGlobalSession(long sessionId, int sessionTimeout) {
    boolean added =
            globalSessionTracker.addSession(sessionId, sessionTimeout);
    if (localSessionsEnabled && added) {
        LOG.info("Adding global session 0x" + Long.toHexString(sessionId));
    }
    return added;
}