private void readRequest() throws IOException {
  zkServer.processPacket(this, incomingBuffer);
}

if (h.getType() == OpCode.auth) {
  LOG.info("got auth packet " + cnxn.getRemoteSocketAddress());
  // 创建安全数据包
  AuthPacket authPacket = new AuthPacket();
  // 将字节缓存中的数据转换为安全数据包对象
  ByteBufferInputStream.byteBuffer2Record(incomingBuffer, authPacket);
  // 获取安全方案
  String scheme = authPacket.getScheme();
  // 根据安全方案获取安全验证器
  AuthenticationProvider ap = ProviderRegistry.getProvider(scheme);
  // 设置安全状态码，第一次设置为安全认证失败
  Code authReturn = KeeperException.Code.AUTHFAILED;
  if (ap != null) {
    try {
      // 从安全验证器中获取状态码
      authReturn = ap.handleAuthentication(cnxn, authPacket.getAuth());
    } catch (RuntimeException e) {
      LOG.warn("Caught runtime exception from AuthenticationProvider: " + scheme
               + " due to " + e);
      // 安全状态码设置为验证失败
      authReturn = KeeperException.Code.AUTHFAILED;
    }
  }
  // 如果是安全状态码是OK则写出响应
  if (authReturn == KeeperException.Code.OK) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Authentication succeeded for scheme: " + scheme);
    }
    LOG.info("auth success " + cnxn.getRemoteSocketAddress());
    ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.OK.intValue());
    cnxn.sendResponse(rh, null, null);
  }
  // 安全状态吗非OK的情况下下
  else {
    if (ap == null) {
      LOG.warn("No authentication provider for scheme: " + scheme + " has "
               + ProviderRegistry.listProviders());
    } else {
      LOG.warn("Authentication failed for scheme: " + scheme);
    }
    // 组装响应信息将响应信息发送
    ReplyHeader rh =
      new ReplyHeader(h.getXid(), 0, KeeperException.Code.AUTHFAILED.intValue());
    cnxn.sendResponse(rh, null, null);
    // 发送关闭连接
    cnxn.sendBuffer(ServerCnxnFactory.closeConn);
    // 不再接收数据
    cnxn.disableRecv();
  }
  // 结束处理
  return;
}

private void processSasl(ByteBuffer incomingBuffer, ServerCnxn cnxn,
                         RequestHeader requestHeader) throws IOException {
  LOG.debug("Responding to client SASL token.");
  // 创建 SASL 请求对象
  GetSASLRequest clientTokenRecord = new GetSASLRequest();
  // 将字节缓存中的数据转换为 SASL 请求对象
  ByteBufferInputStream.byteBuffer2Record(incomingBuffer, clientTokenRecord);
  // 从 SASL 请求对象中获取客户端 token
  byte[] clientToken = clientTokenRecord.getToken();
  LOG.debug("Size of client SASL token: " + clientToken.length);
  // 响应token
  byte[] responseToken = null;
  try {
    // 从服务连接对象中获取 SASL 服务
    ZooKeeperSaslServer saslServer = cnxn.zooKeeperSaslServer;
    try {
      // 通过 SASL 服务创建响应token
      responseToken = saslServer.evaluateResponse(clientToken);
      // 判断 SASL 服务是否处理完成
      if (saslServer.isComplete()) {
        // 通过 SASL 服务获取授权id
        String authorizationID = saslServer.getAuthorizationID();
        LOG.info("adding SASL authorization for authorizationID: " + authorizationID);
        // 向服务连接对象中添加安全信息
        cnxn.addAuthInfo(new Id("sasl", authorizationID));
        // 如果系统环境中zookeeper.superUser属性不为空并且授权id数据和系统环境中zookeeper.superUser的属性相同增加安全信息
        if (System.getProperty("zookeeper.superUser") != null &&
            authorizationID.equals(System.getProperty("zookeeper.superUser"))) {
          cnxn.addAuthInfo(new Id("super", ""));
        }
      }
    }
    // 处理过程中出现异常
    catch (SaslException e) {
      LOG.warn("Client {} failed to SASL authenticate: {}",
               cnxn.getRemoteSocketAddress(), e);
      // 允许 SASL 认证失败，客户端不需要 SASL 验证，输出警告日志
      if (shouldAllowSaslFailedClientsConnect() && !shouldRequireClientSaslAuth()) {
        LOG.warn("Maintaining client connection despite SASL authentication failure.");
      } else {
        //  确认异常值
        int error;
        // 需要 SASL 验证
        if (shouldRequireClientSaslAuth()) {
          LOG.warn(
            "Closing client connection due to server requires client SASL authenticaiton,"
            +
            "but client SASL authentication has failed, or client is not configured with SASL "
            +
            "authentication.");
          // 会话关闭 ， SASL 认证失败
          error = Code.SESSIONCLOSEDREQUIRESASLAUTH.intValue();
        } else {
          LOG.warn("Closing client connection due to SASL authentication failure.");
          // 认证失败
          error = Code.AUTHFAILED.intValue();
        }

        // 组装响应
        ReplyHeader replyHeader = new ReplyHeader(requestHeader.getXid(), 0, error);
        // 发送响应
        cnxn.sendResponse(replyHeader, new SetSASLResponse(null), "response");
        // 发送关闭session消息
        cnxn.sendCloseSession();
        // 不再接收数据
        cnxn.disableRecv();
        return;
      }
    }
  } catch (NullPointerException e) {
    LOG.error(
      "cnxn.saslServer is null: cnxn object did not initialize its saslServer properly.");
  }
  if (responseToken != null) {
    LOG.debug("Size of server SASL response: " + responseToken.length);
  }

  // SASL 验证通过，发送通过信息
  ReplyHeader replyHeader = new ReplyHeader(requestHeader.getXid(), 0, Code.OK.intValue());
  Record record = new SetSASLResponse(responseToken);
  cnxn.sendResponse(replyHeader, record, "response");
}

else {
  // 需要sasl验证并且未通过sasl验证
  if (shouldRequireClientSaslAuth() && !hasCnxSASLAuthenticated(cnxn)) {
    // 组装响应信息将响应信息发送
    ReplyHeader replyHeader = new ReplyHeader(h.getXid(), 0,
                                              Code.SESSIONCLOSEDREQUIRESASLAUTH.intValue());
    cnxn.sendResponse(replyHeader, null, "response");
    // 发送session关闭信息
    cnxn.sendCloseSession();
    // 不再接收数据
    cnxn.disableRecv();
  }
  // 不需要sasl验证或者sasl验证通过
  else {
    Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(), h.getType(),
                             incomingBuffer, cnxn.getAuthInfo());
    si.setOwner(ServerCnxn.me);
    // 确认是否是本地session如果是将isLocalSession设置为true
    setLocalSessionFlag(si);
    // 提交请求，进行处理
    submitRequest(si);
  }
}

public void submitRequest(Request si) {
  // 如果首个请求处理器为空
  if (firstProcessor == null) {
    synchronized (this) {
      try {
        // 由于所有请求都传递给请求处理器，因此它应该等待设置请求处理器链。设置后状态将更新为 RUNNING。
        while (state == State.INITIAL) {
          wait(1000);
        }
      } catch (InterruptedException e) {
        LOG.warn("Unexpected interruption", e);
      }
      if (firstProcessor == null || state != State.RUNNING) {
        throw new RuntimeException("Not started");
      }
    }
  }
  try {
    // 尝试处理 session 相关信息
    touch(si.cnxn);
    // 验证请求类型
    boolean validpacket = Request.isValid(si.type);
    // 请求类型验证通过
    if (validpacket) {
      // 处理请求
      firstProcessor.processRequest(si);
      // 连接对象存在的情况下
      if (si.cnxn != null) {
        // 请求处理数量+1
        incInProcess();
      }
    }
    // 请求类型验证不通过
    else {
      LOG.warn("Received packet at server of unknown type " + si.type);
      // 用于发送错误码响应的请求处理器
      new UnimplementedRequestProcessor().processRequest(si);
    }
  } catch (MissingSessionException e) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Dropping request: " + e.getMessage());
    }
  } catch (RequestProcessorException e) {
    LOG.error("Unable to process request:" + e.getMessage(), e);
  }
}

protected void setupRequestProcessors() {
  RequestProcessor finalProcessor = new FinalRequestProcessor(this);
  RequestProcessor syncProcessor = new SyncRequestProcessor(this,
                                                            finalProcessor);
  ((SyncRequestProcessor) syncProcessor).start();
  firstProcessor = new PrepRequestProcessor(this, syncProcessor);
  ((PrepRequestProcessor) firstProcessor).start();
}

public void processRequest(Request request) {
    submittedRequests.add(request);
}

LinkedBlockingQueue<Request> submittedRequests = new LinkedBlockingQueue<Request>();

@Override
public void run() {
  try {
    // 死循环
    while (true) {
      // 从成员变量submittedRequests中获取一个请求
      Request request = submittedRequests.take();
      // 设置跟踪类型为CLIENT_REQUEST_TRACE_MASK
      long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
      // 如果请求类型是ping，将跟踪类型设置为CLIENT_PING_TRACE_MASK
      if (request.type == OpCode.ping) {
        traceMask = ZooTrace.CLIENT_PING_TRACE_MASK;
      }
      if (LOG.isTraceEnabled()) {
        ZooTrace.logRequest(LOG, traceMask, 'P', request, "");
      }
      // 如果当前请求和死亡请求相同结束死循环停止工作
      if (Request.requestOfDeath == request) {
        break;
      }
      // 处理请求
      pRequest(request);
    }
  } catch (RequestProcessorException e) {
    if (e.getCause() instanceof XidRolloverException) {
      LOG.info(e.getCause().getMessage());
    }
    // 处理异常
    handleException(this.getName(), e);
  } catch (Exception e) {
    handleException(this.getName(), e);
  }
  LOG.info("PrepRequestProcessor exited loop!");
}

private List<ACL> fixupACL(String path, List<Id> authInfo, List<ACL> acls)
  throws KeeperException.InvalidACLException {
  // 将参数 acls 中重复的数据进行过滤
  List<ACL> uniqacls = removeDuplicates(acls);
  // 返回结果值存储容器
  LinkedList<ACL> rv = new LinkedList<ACL>();
  // 如果过滤后的ACL数量为空将抛出异常
  if (uniqacls == null || uniqacls.size() == 0) {
    throw new KeeperException.InvalidACLException(path);
  }
  // 遍历过滤后的acl集合
  for (ACL a : uniqacls) {
    LOG.debug("Processing ACL: {}", a);
    // 如果当前元素为空抛出异常
    if (a == null) {
      throw new KeeperException.InvalidACLException(path);
    }
    // 提取id
    Id id = a.getId();
    // id 为空或者id中存储的方案为空抛出异常
    if (id == null || id.getScheme() == null) {
      throw new KeeperException.InvalidACLException(path);
    }
    // 如果id中存储的方案是world并且id中的id变量为anyone加入到结果集合中
    if (id.getScheme().equals("world") && id.getId().equals("anyone")) {
      rv.add(a);
    }
    // 如果id中存储的方案是auth则进行验证逻辑
    else if (id.getScheme().equals("auth")) {
      // 身份认证通过标记
      boolean authIdValid = false;
      for (Id cid : authInfo) {
        // 根据安全方案获取安全验证器
        AuthenticationProvider ap =
          ProviderRegistry.getProvider(cid.getScheme());

        if (ap == null) {
          LOG.error("Missing AuthenticationProvider for "
                    + cid.getScheme());
        }
        // 通过身份验证
        else if (ap.isAuthenticated()) {
          authIdValid = true;
          rv.add(new ACL(a.getPerms(), cid));
        }
      }
      // 是否认证通过标记为false抛出异常
      if (!authIdValid) {
        throw new KeeperException.InvalidACLException(path);
      }
    }
    else {
      // 根据安全方案获取安全验证器
      AuthenticationProvider ap = ProviderRegistry.getProvider(id.getScheme());
      // 安全验证器为空或者id验证不通过抛出异常
      if (ap == null || !ap.isValid(id.getId())) {
        throw new KeeperException.InvalidACLException(path);
      }
      rv.add(a);
    }
  }
  return rv;
}

private ChangeRecord getRecordForPath(String path) throws KeeperException.NoNodeException {
  ChangeRecord lastChange = null;
  // 锁住未完成的变化记录
  synchronized (zks.outstandingChanges) {
    // 从outstandingChangesForPath对象中获取变化档案
    lastChange = zks.outstandingChangesForPath.get(path);
    if (lastChange == null) {
      // 从zk数据库中获取path对应的数据节点
      DataNode n = zks.getZKDatabase().getNode(path);
      // 如果数据节点不为空
      if (n != null) {
        Set<String> children;
        synchronized (n) {
          // 获取当前节点的子节点
          children = n.getChildren();
        }
        // 重写变化档案
        lastChange = new ChangeRecord(-1, path, n.stat, children.size(),
                                      zks.getZKDatabase().aclForNode(n));
      }
    }
  }
  if (lastChange == null || lastChange.stat == null) {
    throw new KeeperException.NoNodeException(path);
  }
  // 返回
  return lastChange;
}

final Deque<ChangeRecord> outstandingChanges = new ArrayDeque<>();
final HashMap<String, ChangeRecord> outstandingChangesForPath = new HashMap<String, ChangeRecord>();

private void addChangeRecord(ChangeRecord c) {
  synchronized (zks.outstandingChanges) {
    zks.outstandingChanges.add(c);
    zks.outstandingChangesForPath.put(c.path, c);
  }
}

static void checkACL(ZooKeeperServer zks, List<ACL> acl, int perm,
                     List<Id> ids) throws KeeperException.NoAuthException {
  // 跳过acl认证
  if (skipACL) {
    return;
  }
  if (LOG.isDebugEnabled()) {
    LOG.debug("Permission requested: {} ", perm);
    LOG.debug("ACLs for node: {}", acl);
    LOG.debug("Client credentials: {}", ids);
  }
  // acl集合为空或acl数量为空
  if (acl == null || acl.size() == 0) {
    return;
  }
  // 处理id集合
  for (Id authId : ids) {
    // 如果id集合中的元素中有一个方案是super返回
    if (authId.getScheme().equals("super")) {
      return;
    }
  }
  // 处理acl集合
  for (ACL a : acl) {
    Id id = a.getId();
    if ((a.getPerms() & perm) != 0) {
      // 方案是world并且id数据为anyone就放行
      if (id.getScheme().equals("world")
          && id.getId().equals("anyone")) {
        return;
      }
      // 根据安全方案获取安全验证器
      AuthenticationProvider ap = ProviderRegistry.getProvider(id
                                                               .getScheme());
      // 安全验证器不为空
      if (ap != null) {
        // 处理参数ids
        for (Id authId : ids) {
          // 当前id中的方案是否和acl中id的方案相同
          // 安全认证器对当前id和acl中的id进行匹配判断
          if (authId.getScheme().equals(id.getScheme())
              && ap.matches(authId.getId(), id.getId())) {
            return;
          }
        }
      }
    }
  }
  // 抛出异常
  throw new KeeperException.NoAuthException();
}

private void validatePath(String path, long sessionId) throws BadArgumentsException {
  try {
    PathUtils.validatePath(path);
  } catch (IllegalArgumentException ie) {
    LOG.info("Invalid path {} with session 0x{}, reason: {}",
             path, Long.toHexString(sessionId), ie.getMessage());
    throw new BadArgumentsException(path);
  }
}

private void pRequest2TxnCreate(int type, Request request, Record record, boolean deserialize)
  throws IOException, KeeperException {
  // 是否需要反序列化，如果需要则将请求转换到record中
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, record);
  }

  // 标记，该标记表示了创建类型
  int flags;
  // 节点地址
  String path;
  // 权限信息
  List<ACL> acl;
  // 数据信息
  byte[] data;
  // 过期时间
  long ttl;
  // 类型为创建具备过期时间的节点
  if (type == OpCode.createTTL) {
    CreateTTLRequest createTtlRequest = (CreateTTLRequest) record;
    flags = createTtlRequest.getFlags();
    path = createTtlRequest.getPath();
    acl = createTtlRequest.getAcl();
    data = createTtlRequest.getData();
    ttl = createTtlRequest.getTtl();
  }
  else {
    CreateRequest createRequest = (CreateRequest) record;
    flags = createRequest.getFlags();
    path = createRequest.getPath();
    acl = createRequest.getAcl();
    data = createRequest.getData();
    ttl = -1;
  }
  // 将 flags 转换为创建类型
  CreateMode createMode = CreateMode.fromFlag(flags);
  // 验证创建请求是否合法
  validateCreateRequest(path, createMode, request, ttl);
  // 验证创建地址
  String parentPath = validatePathForCreate(path, request.sessionId);

  // 处理ACL
  List<ACL> listACL = fixupACL(path, request.authInfo, acl);
  // 获取父节点的数据记录
  ChangeRecord parentRecord = getRecordForPath(parentPath);

  // 验证ACL
  checkACL(zks, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo);
  // 获取父节点的cversion(创建版本号)
  int parentCVersion = parentRecord.stat.getCversion();
  // 是否是顺序创建
  if (createMode.isSequential()) {
    path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);
  }
  // 验证路径是否合法
  validatePath(path, request.sessionId);
  try {
    // 如果当前路径已经存在相关档案抛出异常
    if (getRecordForPath(path) != null) {
      throw new KeeperException.NodeExistsException(path);
    }
  } catch (KeeperException.NoNodeException e) {
  }
  // 是否是临时节点
  boolean ephemeralParent =
    EphemeralType.get(parentRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL;
  if (ephemeralParent) {
    throw new KeeperException.NoChildrenForEphemeralsException(path);
  }
  // 计算新的创建版本号
  int newCversion = parentRecord.stat.getCversion() + 1;
  // 如果操作类型是创建容器则将txn设置为CreateContainerTxn
  if (type == OpCode.createContainer) {
    request.setTxn(new CreateContainerTxn(path, data, listACL, newCversion));
  }
  // 如果操作类型是创建过期节点则将txn设置为CreateTTLTxn
  else if (type == OpCode.createTTL) {
    request.setTxn(new CreateTTLTxn(path, data, listACL, newCversion, ttl));
  }
  // 其他创建相关的操作类型将txn设置为CreateTxn
  else {
    request.setTxn(new CreateTxn(path, data, listACL, createMode.isEphemeral(),
                                 newCversion));
  }
  StatPersisted s = new StatPersisted();
  if (createMode.isEphemeral()) {
    s.setEphemeralOwner(request.sessionId);
  }
  // 拷贝父节点数据
  parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
  // 父节点中子节点数量加一
  parentRecord.childCount++;
  // 设置创建版本号
  parentRecord.stat.setCversion(newCversion);
  // 添加变化档案
  addChangeRecord(parentRecord);
  addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, s, 0, listACL));
}

private void validateCreateRequest(String path,
                                   CreateMode createMode,
                                   Request request,
                                   long ttl)
  throws KeeperException {
  // 创建模型是否是TTL的
  // 是否启用临时节点的拓展属性
  if (createMode.isTTL() && !EphemeralType.extendedEphemeralTypesEnabled()) {
    throw new KeeperException.UnimplementedException();
  }
  try {
    // 验证TTL是否合法
    EphemeralType.validateTTL(createMode, ttl);
  } catch (IllegalArgumentException e) {
    throw new BadArgumentsException(path);
  }
  // 创建模型属于临时的
  if (createMode.isEphemeral()) {
    // 请求中是否携带异常
    if (request.getException() != null) {
      throw request.getException();
    }
    // 会话跟踪器验证会话
    zks.sessionTracker.checkGlobalSession(request.sessionId,
                                          request.getOwner());
  }
  else {
    // 会话跟踪器验证会话
    zks.sessionTracker.checkSession(request.sessionId,
                                    request.getOwner());
  }
}

private String validatePathForCreate(String path, long sessionId)
  throws BadArgumentsException {
  int lastSlash = path.lastIndexOf('/');
  if (lastSlash == -1 || path.indexOf('\0') != -1 || failCreate) {
    LOG.info("Invalid path %s with session 0x%s",
             path, Long.toHexString(sessionId));
    throw new KeeperException.BadArgumentsException(path);
  }
  return path.substring(0, lastSlash);
}

case OpCode.deleteContainer: {
  // 提取请求中的路径
  String path = new String(request.request.array());
  // 提取父路径
  String parentPath = getParentPathAndValidate(path);
  // 父节点档案信息
  ChangeRecord parentRecord = getRecordForPath(parentPath);
  // 当前节点档案信息
  ChangeRecord nodeRecord = getRecordForPath(path);
  // 当前节点存在子节点抛出异常
  if (nodeRecord.childCount > 0) {
    throw new KeeperException.NotEmptyException(path);
  }
  if (EphemeralType.get(nodeRecord.stat.getEphemeralOwner())
      == EphemeralType.NORMAL) {
    throw new KeeperException.BadVersionException(path);
  }
  // 设置 txn
  request.setTxn(new DeleteTxn(path));
  // 拷贝父节点数据
  parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
  // 父节点的子节点数量减一
  parentRecord.childCount--;
  // 添加变化档案
  addChangeRecord(parentRecord);
  addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, null, -1, null));
  break;
}

case OpCode.delete:
  // 通过sessionTracker验证session
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // 转换为删除请求
  DeleteRequest deleteRequest = (DeleteRequest) record;
  // 如果需要反序列化请求对象中的信息反序列化到删除请求中
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, deleteRequest);
  }
  // 获取需要删除的路径
  String path = deleteRequest.getPath();
  // 获取删除路径的父路径
  String parentPath = getParentPathAndValidate(path);
  // 获取父路径的档案
  ChangeRecord parentRecord = getRecordForPath(parentPath);
  // 获取当前路径的档案
  ChangeRecord nodeRecord = getRecordForPath(path);
  // 验证ACL
  checkACL(zks, parentRecord.acl, ZooDefs.Perms.DELETE, request.authInfo);
  // 验证版本号
  checkAndIncVersion(nodeRecord.stat.getVersion(), deleteRequest.getVersion(), path);
  // 当前节点的子节点数量大于0抛出异常
  if (nodeRecord.childCount > 0) {
    throw new KeeperException.NotEmptyException(path);
  }
  // 设置txn
  request.setTxn(new DeleteTxn(path));
  // 拷贝父节点
  parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
  // 父节点的子节点数量减一
  parentRecord.childCount--;
  // 加入变化档案
  addChangeRecord(parentRecord);
  addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, null, -1, null));
  break;

private static int checkAndIncVersion(int currentVersion, int expectedVersion, String path)
  throws KeeperException.BadVersionException {
  if (expectedVersion != -1 && expectedVersion != currentVersion) {
    throw new KeeperException.BadVersionException(path);
  }
  return currentVersion + 1;
}

case OpCode.setData: 
  // 通过sessionTracker验证session
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // 转换为设置数据请求
  SetDataRequest setDataRequest = (SetDataRequest) record;
  // 如果需要反序列化请求对象中的信息反序列化到设置数据请求中
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, setDataRequest);
  }
  // 获取当前操作路径
  path = setDataRequest.getPath();
  // 验证当前操作路径
  validatePath(path, request.sessionId);
  // 获取当前操作路径对应的档案
  nodeRecord = getRecordForPath(path);
  // 验证权限
  checkACL(zks, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo);
  // 计算新的版本
  int newVersion = checkAndIncVersion(nodeRecord.stat.getVersion(),
                                      setDataRequest.getVersion(), path);

  // 设置txn
  request.setTxn(new SetDataTxn(path, setDataRequest.getData(), newVersion));
  // 拷贝当前节点的档案
  nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
  // 设置版本号
  nodeRecord.stat.setVersion(newVersion);
  // 添加变化档案
  addChangeRecord(nodeRecord);
  break;

case OpCode.reconfig:
  // 如果zk服务没有启动配置重载将抛出异常
  if (!zks.isReconfigEnabled()) {
    LOG.error("Reconfig operation requested but reconfig feature is disabled.");
    throw new KeeperException.ReconfigDisabledException();
  }

  // 如果跳过acl会输出警告日志
  if (skipACL) {
    LOG.warn("skipACL is set, reconfig operation will skip ACL checks!");
  }

  // 通过sessionTracker验证session
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // 转换为重载配置请求
  ReconfigRequest reconfigRequest = (ReconfigRequest) record;
  // 确认leader服务
  LeaderZooKeeperServer lzks;
  try {
    // 如果当前zk服务无法转换为LeaderZooKeeperServer将抛出异常
    lzks = (LeaderZooKeeperServer) zks;
  } catch (ClassCastException e) {
    throw new KeeperException.UnimplementedException();
  }
  // 获取最后一个验证器
  QuorumVerifier lastSeenQV = lzks.self.getLastSeenQuorumVerifier();
  // 最后一个验证器的版本和leader服务中的验证器版本不相等将抛出异常
  if (lastSeenQV.getVersion() != lzks.self.getQuorumVerifier().getVersion()) {
    throw new KeeperException.ReconfigInProgress();
  }
  // 获取配置id
  long configId = reconfigRequest.getCurConfigId();

  // 配置id为-1
  // 配置id不等于最后一个验证器的版本
  if (configId != -1 && configId != lzks.self.getLastSeenQuorumVerifier()
      .getVersion()) {
    String msg = "Reconfiguration from version " + configId
      + " failed -- last seen version is " +
      lzks.self.getLastSeenQuorumVerifier().getVersion();
    throw new KeeperException.BadVersionException(msg);
  }

  // 获取新成员(配置数据)
  String newMembers = reconfigRequest.getNewMembers();

  // 新成员不为空
  if (newMembers != null) { 
    LOG.info("Non-incremental reconfig");
    newMembers = newMembers.replaceAll(",", "\n");

    try {
      // 读取新成员数据
      Properties props = new Properties();
      props.load(new StringReader(newMembers));
      // 解析配置
      request.qv = QuorumPeerConfig.parseDynamicConfig(props,
                                                       lzks.self.getElectionType(), true, false);
      request.qv.setVersion(request.getHdr().getZxid());
    } catch (IOException | ConfigException e) {
      throw new KeeperException.BadArgumentsException(e.getMessage());
    }
  } else {
    LOG.info("Incremental reconfig");

    // 求需要进入集群的服务
    List<String> joiningServers = null;
    String joiningServersString = reconfigRequest.getJoiningServers();
    if (joiningServersString != null) {
      joiningServers = StringUtils.split(joiningServersString, ",");
    }

    // 求需要离开的服务
    List<String> leavingServers = null;
    String leavingServersString = reconfigRequest.getLeavingServers();
    if (leavingServersString != null) {
      leavingServers = StringUtils.split(leavingServersString, ",");
    }

    // 类型不同抛出异常
    if (!(lastSeenQV instanceof QuorumMaj)) {
      String msg =
        "Incremental reconfiguration requested but last configuration seen has a non-majority quorum system";
      LOG.warn(msg);
      throw new KeeperException.BadArgumentsException(msg);
    }
    // 服务集合
    Map<Long, QuorumServer> nextServers =
      new HashMap<Long, QuorumServer>(lastSeenQV.getAllMembers());
    try {
      // 如果需要离开的服务变量存在则将nextServers中对应的数据删除
      if (leavingServers != null) {
        for (String leaving : leavingServers) {
          long sid = Long.parseLong(leaving);
          nextServers.remove(sid);
        }
      }
      // 如果需要进入的服务存在则将数据加入到nextServers中
      if (joiningServers != null) {
        for (String joiner : joiningServers) {
          String[] parts =
            StringUtils.split(joiner, "=").toArray(new String[0]);
          if (parts.length != 2) {
            throw new KeeperException.BadArgumentsException(
              "Wrong format of server string");
          }
          Long sid = Long.parseLong(
            parts[0].substring(parts[0].lastIndexOf('.') + 1));
          QuorumServer qs = new QuorumServer(sid, parts[1]);
          if (qs.clientAddr == null || qs.electionAddr == null
              || qs.addr == null) {
            throw new KeeperException.BadArgumentsException(
              "Wrong format of server string - each server should have 3 ports specified");
          }
          for (QuorumServer nqs : nextServers.values()) {
            if (qs.id == nqs.id) {
              continue;
            }
            qs.checkAddressDuplicate(nqs);
          }

          nextServers.remove(qs.id);
          nextServers.put(qs.id, qs);
        }
      }
    } catch (ConfigException e) {
      throw new KeeperException.BadArgumentsException("Reconfiguration failed");
    }
    // 设置请求参数qv
    request.qv = new QuorumMaj(nextServers);
    request.qv.setVersion(request.getHdr().getZxid());
  }
  // 如果仲裁配置启用并且投票成员数量小于2，抛出异常
  if (QuorumPeerConfig.isStandaloneEnabled()
      && request.qv.getVotingMembers().size() < 2) {
    String msg =
      "Reconfig failed - new configuration must include at least 2 followers";
    LOG.warn(msg);
    throw new KeeperException.BadArgumentsException(msg);
  }
  // 如果投票成员数量小于1抛出异常
  else if (request.qv.getVotingMembers().size() < 1) {
    String msg =
      "Reconfig failed - new configuration must include at least 1 follower";
    LOG.warn(msg);
    throw new KeeperException.BadArgumentsException(msg);
  }

  // 仲裁数据是否同步，如果没有同步抛出异常
  if (!lzks.getLeader().isQuorumSynced(request.qv)) {
    String msg2 =
      "Reconfig failed - there must be a connected and synced quorum in new configuration";
    LOG.warn(msg2);
    throw new KeeperException.NewConfigNoQuorum();
  }

  // 获取配置节点的档案
  nodeRecord = getRecordForPath(ZooDefs.CONFIG_NODE);
  // 验证权限
  checkACL(zks, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo);
  // 设置txn
  request.setTxn(
    new SetDataTxn(ZooDefs.CONFIG_NODE, request.qv.toString().getBytes(), -1));
  nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
  nodeRecord.stat.setVersion(-1);
  addChangeRecord(nodeRecord);
  break;

case OpCode.setACL:
  // 通过sessionTracker验证session
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // 类型转换
  SetACLRequest setAclRequest = (SetACLRequest) record;
  // 反序列化
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, setAclRequest);
  }
  // 获取操作路径
  path = setAclRequest.getPath();
  // 验证路径
  validatePath(path, request.sessionId);
  // 计算ACL
  List<ACL> listACL = fixupACL(path, request.authInfo, setAclRequest.getAcl());
  // 获取当前路径对应的档案
  nodeRecord = getRecordForPath(path);
  // 校验ACL
  checkACL(zks, nodeRecord.acl, ZooDefs.Perms.ADMIN, request.authInfo);
  // 计算新版本号
  newVersion = checkAndIncVersion(nodeRecord.stat.getAversion(),
                                  setAclRequest.getVersion(), path);
  // 设置txn
  request.setTxn(new SetACLTxn(path, listACL, newVersion));
  // 拷贝当前路径对应的档案
  nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
  nodeRecord.stat.setAversion(newVersion);
  addChangeRecord(nodeRecord);
  break;

case OpCode.createSession:
  request.request.rewind();
	// 超时时间
  int to = request.request.getInt();
  request.setTxn(new CreateSessionTxn(to));
  request.request.rewind();
	// 不同状态设置sessionId
  if (request.isLocalSession()) {
    zks.sessionTracker.addSession(request.sessionId, to);
  } else {
    zks.sessionTracker.addGlobalSession(request.sessionId, to);
  }
  zks.setOwner(request.sessionId, request.getOwner());
  break;

case OpCode.closeSession:
  // 根据sessionId获取对应的临时路径
  Set<String> es = zks.getZKDatabase()
    .getEphemerals(request.sessionId);
  // 锁
  synchronized (zks.outstandingChanges) {
    // 循环outstandingChanges对象
    for (ChangeRecord c : zks.outstandingChanges) {
      // 如果状态为空则
      if (c.stat == null) {
        // Doing a delete
        // 在临时路径集合中删除
        es.remove(c.path);
      }
      // 如果状态的临时拥有者和当前sessionId相同则将其加入到临时路径集合中
      else if (c.stat.getEphemeralOwner() == request.sessionId) {
        es.add(c.path);
      }
    }
    // 循环进行删除标记
    for (String path2Delete : es) {
      addChangeRecord(
        new ChangeRecord(request.getHdr().getZxid(), path2Delete, null, 0,
                         null));
    }

    // 关闭sessionId
    zks.sessionTracker.setSessionClosing(request.sessionId);
  }
  break;

case OpCode.check:
  // 通过sessionTracker验证session
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // 类型转换
  CheckVersionRequest checkVersionRequest = (CheckVersionRequest) record;
  // 反序列化
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, checkVersionRequest);
  }
  // 提取需要操作路径
  path = checkVersionRequest.getPath();
  // 验证路径
  validatePath(path, request.sessionId);
  // 获取操作路径对应的档案
  nodeRecord = getRecordForPath(path);
  // 验证acl
  checkACL(zks, nodeRecord.acl, ZooDefs.Perms.READ, request.authInfo);
  // 设置txn
  request.setTxn(
    new CheckVersionTxn(path, checkAndIncVersion(nodeRecord.stat.getVersion(),
                                                 checkVersionRequest.getVersion(), path)));
  break;

case OpCode.multi:{
  // 创建多事务请求对象
  MultiTransactionRecord multiRequest = new MultiTransactionRecord();
  try {
    // 请求对象转换为多事务请求对象
    ByteBufferInputStream.byteBuffer2Record(request.request, multiRequest);
  } catch (IOException e) {
    // 异常设置头信息
    request.setHdr(
      new TxnHeader(request.sessionId, request.cxid, zks.getNextZxid(),
                    Time.currentWallTime(), OpCode.multi));
    throw e;
  }
  List<Txn> txns = new ArrayList<Txn>();
  // 获取下一个zxid
  long zxid = zks.getNextZxid();
  // 异常记录变量
  KeeperException ke = null;
  // 将多事务请求对象转换为map结构，用于后续回滚操作
  Map<String, ChangeRecord> pendingChanges = getPendingChanges(multiRequest);

  // 循环处理多事务请求对象
  for (Op op : multiRequest) {
    // 从单个事务中获取请求信息
    Record subrequest = op.toRequestRecord();
    // 操作类型
    int type;
    // 档案信息
    Record txn;
    // 如果异常记录变量不为空设置异常相关信息
    if (ke != null) {
      type = OpCode.error;
      txn = new ErrorTxn(Code.RUNTIMEINCONSISTENCY.intValue());
    }
    else {
      try {
        // 执行核心请求转事务逻辑
        pRequest2Txn(op.getType(), zxid, request, subrequest, false);
        // 记录操作类型
        type = request.getHdr().getType();
        // 记录txn
        txn = request.getTxn();
      }
      // 异常相关信息记录
      catch (KeeperException e) {

        ke = e;
        type = OpCode.error;
        txn = new ErrorTxn(e.code().intValue());

        if (e.code().intValue() > Code.APIERROR.intValue()) {
          LOG.info(
            "Got user-level KeeperException when processing {} aborting"
            +
            " remaining multi ops. Error Path:{} Error:{}",
            request.toString(), e.getPath(), e.getMessage());
        }

        request.setException(e);
        // 回滚操作
        rollbackPendingChanges(zxid, pendingChanges);
      }
    }
    // 输出档案处理
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
    txn.serialize(boa, "request");
    ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());

    txns.add(new Txn(type, bb.array()));
  }

  request.setHdr(new TxnHeader(request.sessionId, request.cxid, zxid,
                               Time.currentWallTime(), request.type));
  request.setTxn(new MultiTxn(txns));

  break;
}

void rollbackPendingChanges(long zxid, Map<String, ChangeRecord> pendingChangeRecords) {
  // 锁
  synchronized (zks.outstandingChanges) {
    // 遍历outstandingChanges变量
    Iterator<ChangeRecord> iter = zks.outstandingChanges.descendingIterator();
    while (iter.hasNext()) {
      ChangeRecord c = iter.next();
      // 如果操作的档案和当前档案相同
      if (c.zxid == zxid) {
        iter.remove();
        zks.outstandingChangesForPath.remove(c.path);
      }
      else {
        break;
      }
    }

    // outstandingChanges变量为空
    if (zks.outstandingChanges.isEmpty()) {
      return;
    }

    // 获取第一个zxid
    long firstZxid = zks.outstandingChanges.peek().zxid;

    // 小于第一个zxid的都跳过处理，大于等于的将其加入到outstandingChangesForPath变量中
    for (ChangeRecord c : pendingChangeRecords.values()) {
      if (c.zxid < firstZxid) {
        continue;
      }
      zks.outstandingChangesForPath.put(c.path, c);
    }
  }
}

public LeaderRequestProcessor(LeaderZooKeeperServer zks,
                              RequestProcessor nextProcessor) {
  this.lzks = zks;
  this.nextProcessor = nextProcessor;
}

@Override
protected void setupRequestProcessors() {
  RequestProcessor finalProcessor = new FinalRequestProcessor(this);
  RequestProcessor toBeAppliedProcessor =
    new Leader.ToBeAppliedRequestProcessor(finalProcessor, getLeader());
  commitProcessor = new CommitProcessor(toBeAppliedProcessor,
                                        Long.toString(getServerId()), false,
                                        getZooKeeperServerListener());
  commitProcessor.start();
  ProposalRequestProcessor proposalProcessor = new ProposalRequestProcessor(this,
                                                                            commitProcessor);
  proposalProcessor.initialize();
  prepRequestProcessor = new PrepRequestProcessor(this, proposalProcessor);
  prepRequestProcessor.start();
  firstProcessor = new LeaderRequestProcessor(this, prepRequestProcessor);

  setupContainerManager();
}

@Override
public void processRequest(Request request)
  throws RequestProcessorException {
  // 升级后的请求对象
  Request upgradeRequest = null;
  try {
    // 验证session确认是否需要升级，如果需要则进行升级操作
    upgradeRequest = lzks.checkUpgradeSession(request);
  } catch (KeeperException ke) {
    if (request.getHdr() != null) {
      LOG.debug("Updating header");
      request.getHdr().setType(OpCode.error);
      request.setTxn(new ErrorTxn(ke.code().intValue()));
    }
    request.setException(ke);
    LOG.info("Error creating upgrade request " + ke.getMessage());
  } catch (IOException ie) {
    LOG.error("Unexpected error in upgrade", ie);
  }
  // 如果升级后的请求对象不为空
  if (upgradeRequest != null) {
    nextProcessor.processRequest(upgradeRequest);
  }
  nextProcessor.processRequest(request);
}

@Override
public void processRequest(Request request) {
  if (stopped) {
    return;
  }
  if (LOG.isDebugEnabled()) {
    LOG.debug("Processing request:: " + request);
  }
  queuedRequests.add(request);
  if (!isWaitingForCommit()) {
    wakeup();
  }
}

@Override
public void run() {
  Request request;
  try {
    // 没有停止的情况下死循环
    while (!stopped) {
      // 锁
      synchronized (this) {
        // 需要等待的条件：
        // 1. 未停止
        // 2.1 queuedRequests集合为空
        // 2.2 nextPending 中存在数据
        // 2.3 currentlyCommitting 中存在数据
        // 3.1 committedRequests为空
        // 3.2 numRequestsProcessing数量不为0
        while (
          !stopped &&
          ((queuedRequests.isEmpty() || isWaitingForCommit()
            || isProcessingCommit()) &&
           (committedRequests.isEmpty()
            || isProcessingRequest()))) {

          wait();
        }
      }
      // 处理queuedRequests中的请求
      // 死循环条件
      // 1. 未停止工作
      // 2. nextPending 中存在数据
      // 3. currentlyCommitting 中存在数据
      // 4. queuedRequests 中获取一个请求对象，该请求对象不为空
      while (!stopped && !isWaitingForCommit() &&
             !isProcessingCommit() &&
             (request = queuedRequests.poll()) != null) {
        // 判断是否需要提交
        if (needCommit(request)) {
          nextPending.set(request);
        }
        else {
          // 发送给下一个请求处理器处理
          sendToNextProcessor(request);
        }
      }
      // 处理已提交的请求
      processCommitted();
    }
  } catch (Throwable e) {
    handleException(this.getName(), e);
  }
  LOG.info("CommitProcessor exited loop!");
}

private void sendToNextProcessor(Request request) {
  numRequestsProcessing.incrementAndGet();
  workerPool.schedule(new CommitWorkRequest(request), request.sessionId);
}

public void schedule(WorkRequest workRequest, long id) {
  // 判断是否停止处理
  if (stopped) {
    // 清空请求
    workRequest.cleanup();
    return;
  }

  // 转换请求
  ScheduledWorkRequest scheduledWorkRequest =
    new ScheduledWorkRequest(workRequest);
  // 获取任务执行器数量
  int size = workers.size();
  // 任务执行器数量大于0
  if (size > 0) {
    try {
      int workerNum = ((int) (id % size) + size) % size;
      // 取出任务执行器执行
      ExecutorService worker = workers.get(workerNum);
      worker.execute(scheduledWorkRequest);
    } catch (RejectedExecutionException e) {
      LOG.warn("ExecutorService rejected execution", e);
      workRequest.cleanup();
    }
  }
  else {
    scheduledWorkRequest.run();
  }
}

private class ScheduledWorkRequest implements Runnable {
  private final WorkRequest workRequest;

  ScheduledWorkRequest(WorkRequest workRequest) {
    this.workRequest = workRequest;
  }

  @Override
  public void run() {
    try {
      if (stopped) {
        workRequest.cleanup();
        return;
      }
      workRequest.doWork();
    } catch (Exception e) {
      LOG.warn("Unexpected exception", e);
      workRequest.cleanup();
    }
  }
}

public void doWork() throws RequestProcessorException {
    try {
        nextProcessor.processRequest(request);
    } finally {
        currentlyCommitting.compareAndSet(request, null);
        if (numRequestsProcessing.decrementAndGet() == 0) {
            if (!queuedRequests.isEmpty() ||
                    !committedRequests.isEmpty()) {
                wakeup();
            }
        }
    }
}

protected void processCommitted() {
  // 请求对象
  Request request;

  // 满足条件执行
  // 1. 没有停止工作
  // 2. numRequestsProcessing数量为0
  // 3. 从committedRequests中取出一个数据不为空
  if (!stopped && !isProcessingRequest() &&
      (committedRequests.peek() != null)) {
    // 1. nextPending中存在数据
    // 2. queuedRequests不为空
    if (!isWaitingForCommit() && !queuedRequests.isEmpty()) {
      return;
    }
    // 取出请求处理
    request = committedRequests.poll();
    // 从 nextPending 中取出下一个请求
    Request pending = nextPending.get();
    // 1. 下一个请求不为空
    // 2. 下一个请求与当前请求的sessionId相同
    // 3. 下一个请求与当前请求的cxid相同
    if (pending != null &&
        pending.sessionId == request.sessionId &&
        pending.cxid == request.cxid) {
      pending.setHdr(request.getHdr());
      pending.setTxn(request.getTxn());
      pending.zxid = request.zxid;
      currentlyCommitting.set(pending);
      nextPending.set(null);
      sendToNextProcessor(pending);
    }
    else {
      currentlyCommitting.set(request);
      sendToNextProcessor(request);
    }
  }
}

public void processRequest(Request request) throws RequestProcessorException {
  if (request instanceof LearnerSyncRequest) {
    zks.getLeader().processSync((LearnerSyncRequest) request);
  } else {
    nextProcessor.processRequest(request);
    if (request.getHdr() != null) {
      try {
        // 发送给其他服务
        zks.getLeader().propose(request);
      } catch (XidRolloverException e) {
        throw new RequestProcessorException(e.getMessage(), e);
      }
      syncProcessor.processRequest(request);
    }
  }
}

@Override
public void run() {
  try {
    // 日志计数器
    int logCount = 0;
    // 随机快照数量，避免全量快照日志
    int randRoll = r.nextInt(snapCount / 2);
    while (true) {
      Request si = null;
      // 从queuedRequests集合中取出请求对象
      if (toFlush.isEmpty()) {
        si = queuedRequests.take();
      } else {
        si = queuedRequests.poll();
        if (si == null) {
          flush(toFlush);
          continue;
        }
      }
      // 如果请求对象和死亡请求对象相同则跳出死循环
      if (si == requestOfDeath) {
        break;
      }
      if (si != null) {
        // 向zk数据库追加当前请求
        if (zks.getZKDatabase().append(si)) {
          // 日志计数器累加一
          logCount++;
          // 日志计数器
          if (logCount > (snapCount / 2 + randRoll)) {
            randRoll = r.nextInt(snapCount / 2);
            // roll the log
            // 滚动日志
            zks.getZKDatabase().rollLog();
            // 确认是否需要生成快照
            if (snapInProcess != null && snapInProcess.isAlive()) {
              LOG.warn("Too busy to snap, skipping");
            }
            else {
              snapInProcess = new ZooKeeperThread("Snapshot Thread") {
                public void run() {
                  try {
                    // 生成快照
                    zks.takeSnapshot();
                  } catch (Exception e) {
                    LOG.warn("Unexpected exception", e);
                  }
                }
              };
              // 启动
              snapInProcess.start();
            }
            // 计数器归零
            logCount = 0;
          }
        }
        // 等待写入的请求为空
        else if (toFlush.isEmpty()) {
          if (nextProcessor != null) {
            nextProcessor.processRequest(si);
            if (nextProcessor instanceof Flushable) {
              ((Flushable) nextProcessor).flush();
            }
          }
          continue;
        }
        // 等待写入的请求
        toFlush.add(si);
        // 超过1000个进行写入操作
        if (toFlush.size() > 1000) {
          flush(toFlush);
        }
      }
    }
  } catch (Throwable t) {
    handleException(this.getName(), t);
  } finally {
    running = false;
  }
  LOG.info("SyncRequestProcessor exited!");
}