struct{
    czxid;// create ZXID，即该数据节点被创建时的事务id
    mzxid;// modified ZXID，即该节点最终一次更新时的事务id
    pzxid;// 该节点的子节点列表最后一次修改时的事务id，只有子节点列表变更才会更新pZxid，子节点内容变更不会更新
    ctime;// create time，即该节点的创建时间
    mtime;// modified time，即该节点最后一次的更新时间
    version;// 数据节点内容版本号，节点创建时为0，每更新一次节点内容(不管内容有无变化)该版本号的值增加1
    cversion;//子节点版本号，当前节点的子节点每次变化时值增加1
    aversion;// 节点的ACL版本号，表示该节点ACL信息变更次数
    ephemeralOwner;// 创建该临时节点的会话的sessionId；如果当前节点为持久节点，则ephemeralOwner=0
    dataLength;// 数据节点内容长度
    numChildren;// 当前节点的子节点个数
}

void followLeader() throws InterruptedException {
try {
    ......
    while (this.isRunning()) {
        readPacket(qp);
        processPacket(qp);
    }
    // 如果上面的while循环内出现异常
    // 注意：长时间没有收到Leader的消息也是异常
} catch (Exception e) {
    // 出现异常就退出了while循环
    // 也就结束了Follower的处理流程
}

public void run() {
    while (running) {
        switch (getPeerState()) {
        case FOLLOWING:
            try {
                setFollower(makeFollower(logFactory));
                follower.followLeader();
            } catch (Exception e) {
                ......
            } finally {
                follower.shutdown();
                setFollower(null);
                // 状态更新为LOOKING
                updateServerState();
            }
            break;
            //......
    }
}

public void run() {
    ......
    while (true) {
        qp = new QuorumPacket();
        ia.readRecord(qp, "packet");
        //......
        // 收到Follower的消息后
        // 设置下一个消息的过期时间
        tickOfNextAckDeadline = leader.self.tick.get() + leader.self.syncLimit;
        //......
    }
    //......
}

void lead() throws IOException, InterruptedException {
    //......
    while (true) {
        //......
        // 判断每个每个Follower节点的状态
        // 是否与Leader保持同步
        for (LearnerHandler f : getLearners()) {
            if (f.synced()) {   
                syncedAckSet.addAck(f.getSid());
            }
        }
        //......
    }
    if (!tickSkip && !syncedAckSet.hasAllQuorums()) {
        // 如果失去了大多数Follower节点的认可，就跳出Leader主循环，进入选主流程
        break;
    }
    //......
}

// LearnerHandler::synced()逻辑
// 即判断当前是否已经过了期望得到的Follower的下一个消息的期限：tickOfNextAckDeadline
public boolean synced() {
    return isAlive() && leader.self.tick.get() <= tickOfNextAckDeadline;
}

public static void main(String[] args) {
        QuorumPeerMain main = new QuorumPeerMain();
        try {//初始化主要逻辑
            main.initializeAndRun(args);
        }
　　　　　//...异常捕获
        LOG.info("Exiting normally");
        System.exit(0);
    }

protected void initializeAndRun(String[] args)
        throws ConfigException, IOException
    {
        QuorumPeerConfig config = new QuorumPeerConfig();
        if (args.length == 1) {
            config.parse(args[0]);
        }
        // 启动后台定时任务异步执行清除任务，删除垃圾数据
        // Start and schedule the the purge task
        DatadirCleanupManager purgeMgr = new DatadirCleanupManager(config
                .getDataDir(), config.getDataLogDir(), config
                .getSnapRetainCount(), config.getPurgeInterval());
        purgeMgr.start();
        //判断是集群还是单机
        if (args.length == 1 && config.servers.size() > 0) {
            // 集群
            runFromConfig(config);
        } else {
            LOG.warn("Either no config or no quorum defined in config, running "
                    + " in standalone mode");
            // there is only server in the quorum -- run as standalone
            //单机
            ZooKeeperServerMain.main(args);
        }
    }

public void runFromConfig(QuorumPeerConfig config) throws IOException {
      try {
          ManagedUtil.registerLog4jMBeans();
      } catch (JMException e) {
          LOG.warn("Unable to register log4j JMX control", e);
      }
  
      LOG.info("Starting quorum peer");
      try {// 初始化NIOServerCnxnFactory
          ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
          cnxnFactory.configure(config.getClientPortAddress(),
                                config.getMaxClientCnxns());
         // 逻辑主线程 进行投票，选举
          quorumPeer = getQuorumPeer();
          // 进入一系列的配置
          quorumPeer.setQuorumPeers(config.getServers());
          quorumPeer.setTxnFactory(new FileTxnSnapLog(
                  new File(config.getDataLogDir()),
                  new File(config.getDataDir())));
          quorumPeer.setElectionType(config.getElectionAlg());
          quorumPeer.setMyid(config.getServerId()); //配置 myid
          quorumPeer.setTickTime(config.getTickTime());
          quorumPeer.setInitLimit(config.getInitLimit());
          quorumPeer.setSyncLimit(config.getSyncLimit());
          quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());
          quorumPeer.setCnxnFactory(cnxnFactory);
          quorumPeer.setQuorumVerifier(config.getQuorumVerifier());
          // 为客户端提供写的server 即2181访问端口的访问功能
          quorumPeer.setClientPortAddress(config.getClientPortAddress());
          quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
          quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
          quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
          quorumPeer.setLearnerType(config.getPeerType());
          quorumPeer.setSyncEnabled(config.getSyncEnabled());

          // sets quorum sasl authentication configurations
          quorumPeer.setQuorumSaslEnabled(config.quorumEnableSasl);
          if(quorumPeer.isQuorumSaslAuthEnabled()){
              quorumPeer.setQuorumServerSaslRequired(config.quorumServerRequireSasl);
              quorumPeer.setQuorumLearnerSaslRequired(config.quorumLearnerRequireSasl);
              quorumPeer.setQuorumServicePrincipal(config.quorumServicePrincipal);
              quorumPeer.setQuorumServerLoginContext(config.quorumServerLoginContext);
              quorumPeer.setQuorumLearnerLoginContext(config.quorumLearnerLoginContext);
          }

          quorumPeer.setQuorumCnxnThreadsSize(config.quorumCnxnThreadsSize);
         // 初始化的工作
          quorumPeer.initialize();
          // 启动主线程，QuorumPeer 重写了 Thread.start 方法
          quorumPeer.start();
          quorumPeer.join();//使得线程之间的并行执行变为串行执行
      } catch (InterruptedException e) {
          // warn, but generally this is ok
          LOG.warn("Quorum Peer interrupted", e);
      }
    }

@Override
public synchronized void start() {
        //载入本地DB数据 主要还是epoch
        loadDataBase();
　　　　　//启动ZooKeeperThread线程
        cnxnFactory.start();    
        //启动leader选举线程    
        startLeaderElection();
        super.start();
}

private void loadDataBase() {
        File updating = new File(getTxnFactory().getSnapDir(),
                                 UPDATING_EPOCH_FILENAME);
        try {//载入本地数据
            zkDb.loadDataBase();
            // load the epochs 加载ZXID
            long lastProcessedZxid = zkDb.getDataTree().lastProcessedZxid;
            // 根据zxid的高32位是epoch号，低32位是事务id进行抽离epoch号
            long epochOfZxid = ZxidUtils.getEpochFromZxid(lastProcessedZxid);
            try {//从${data}/version-2/currentEpochs文件中加载当前的epoch号
                currentEpoch = readLongFromFile(CURRENT_EPOCH_FILENAME);
                //从 zxid中提取的epoch比文件里的epoch要大的话，并且没有正在修改epoch
                if (epochOfZxid > currentEpoch && updating.exists()) {
                    setCurrentEpoch(epochOfZxid);//设置位大的epoch
                    if (!updating.delete()) {
                        throw new IOException("Failed to delete " +
                                              updating.toString());
                    }
                }
            } 
　　　　　　　 // ........
            //如果如果还比他大 抛出异常
            if (epochOfZxid > currentEpoch) {
                throw new IOException("The current epoch, " + ZxidUtils.zxidToString(currentEpoch) + ", is older than the last zxid, " + lastProcessedZxid);
            }
            try {//再比较 acceptedEpoch
                acceptedEpoch = readLongFromFile(ACCEPTED_EPOCH_FILENAME);
            }
            // ........
            if (acceptedEpoch < currentEpoch) {
                throw new IOException("The accepted epoch, " + ZxidUtils.zxidToString(acceptedEpoch) + " is less than the current epoch, " + ZxidUtils.zxidToString(currentEpoch));
            }
　　　　　　　// .......
}

synchronized public void startLeaderElection() {
        try { // 根据myid zxid epoch 3个选举参数创建Voto 对象，准备选举
            currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
        } catch(IOException e) {
            RuntimeException re = new RuntimeException(e.getMessage());
            re.setStackTrace(e.getStackTrace());
            throw re;
        }
        for (QuorumServer p : getView().values()) {
            if (p.id == myid) {
                myQuorumAddr = p.addr;
                break;
            }
        }
        if (myQuorumAddr == null) {
            throw new RuntimeException("My id " + myid + " not in the peer list");
        }
        if (electionType == 0) {//如果是这个选举策略，代表 LeaderElection选举策略
            try {//创建 UDP Socket
                udpSocket = new DatagramSocket(myQuorumAddr.getPort());
                responder = new ResponderThread();
                responder.start();
            } catch (SocketException e) {
                throw new RuntimeException(e);
            }
        }//根据类型创建选举算法
        this.electionAlg = createElectionAlgorithm(electionType);
    }

protected Election createElectionAlgorithm(int electionAlgorithm){
        Election le=null;
        // 选择选举策略
        //TODO: use a factory rather than a switch
        switch (electionAlgorithm) {
        case 0:
            le = new LeaderElection(this);
            break;
        case 1:
            le = new AuthFastLeaderElection(this);
            break;
        case 2:
            le = new AuthFastLeaderElection(this, true);
            break;
        case 3://Leader选举IO负责类
            qcm = createCnxnManager();
            QuorumCnxManager.Listener listener = qcm.listener;
            if(listener != null){
                // 启动已绑定端口的选举线程，等待其他服务器连接
                listener.start();
                //基于 TCP的选举算法
                le = new FastLeaderElection(this, qcm);
            } else {
                LOG.error("Null listener when initializing cnx manager");
            }
            break;
        default:
            assert false;
        }
        return le;
    }

public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager){
        this.stop = false;
        this.manager = manager;
        starter(self, manager);
}

// ***********************************************

private void starter(QuorumPeer self, QuorumCnxManager manager) {
        this.self = self;
        proposedLeader = -1;
        proposedZxid = -1;
       // 投票 发送队列 阻塞
        sendqueue = new LinkedBlockingQueue<ToSend>();
        // 投票 接受队列 阻塞
        recvqueue = new LinkedBlockingQueue<Notification>();
        this.messenger = new Messenger(manager);
}

public void run() {
        setName("QuorumPeer" + "[myid=" + getId() + "]" +
                cnxnFactory.getLocalAddress());
        // 省略通过JMX初始化。来监控一些属性的代码
        try {
            // Main loop  主循环
            while (running) {
                switch (getPeerState()) {
                case LOOKING: //LOOKING 状态，则进入选举
                    if (Boolean.getBoolean("readonlymode.enabled")) {
　　　　　　　　　　　　　　 // 创建 ReadOnlyZooKeeperServer，但是不立即启动
                        final ReadOnlyZooKeeperServer roZk = new ReadOnlyZooKeeperServer(
                                logFactory, this,
                                new ZooKeeperServer.BasicDataTreeBuilder(),
                                this.zkDb);
    　　　　　　　　　　　　//通过 Thread 异步解耦
                        Thread roZkMgr = new Thread() {
                            public void run() {
                                try {
                                    // lower-bound grace period to 2 secs
                                    sleep(Math.max(2000, tickTime));
                                    if (ServerState.LOOKING.equals(getPeerState())) {
                                        roZk.startup();
                                    }
　　　　　　　　　　　　　　　　　　　// .......　　
                            }
                        };
                        try {//启动
                            roZkMgr.start();
                            setBCVote(null);
                           // 通过策略模式来决定当前用那个算法选举
                            setCurrentVote(makeLEStrategy().lookForLeader());
                        // .........
                    } else {
                        try {
                            setBCVote(null);
                            setCurrentVote(makeLEStrategy().lookForLeader());
                        //........
                    }
                    break;
//****************************************************************************
                case OBSERVING: // Observing 针对 Observer角色的节点
                    try {
                        LOG.info("OBSERVING");
                        setObserver(makeObserver(logFactory));
                        observer.observeLeader();
                    } catch (Exception e) {
                        LOG.warn("Unexpected exception",e );                        
                    } finally {
                        observer.shutdown();
                        setObserver(null);
                        setPeerState(ServerState.LOOKING);
                    }
                    break;
//*****************************************************************************
                case FOLLOWING:// 从节点状态
                    try {
                        LOG.info("FOLLOWING");
                        setFollower(makeFollower(logFactory));
                        follower.followLeader();
                    } catch (Exception e) {
                        LOG.warn("Unexpected exception",e);
                    } finally {
                        follower.shutdown();
                        setFollower(null);
                        setPeerState(ServerState.LOOKING);
                    }
                    break;
//**********************************************************************
                case LEADING: // leader 节点
                    LOG.info("LEADING");
                    try {
                        setLeader(makeLeader(logFactory));
                        leader.lead();
                        setLeader(null);
                    } catch (Exception e) {
                        LOG.warn("Unexpected exception",e);
                    } finally {
                        if (leader != null) {
                            leader.shutdown("Forcing shutdown");
                            setLeader(null);
                        }
                        setPeerState(ServerState.LOOKING);
                    }
                    break;
                }
            }
        }
　　 // ..........
}

//开始选举 Leader
public Vote lookForLeader() throws InterruptedException {
　　　　 // ...省略一些代码
        try {
            // 收到的投票
            HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();
            // 存储选举结果 
            HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();

            int notTimeout = finalizeWait;
 　　　　　　 // AtomicLong logicalclock = new AtomicLong();
            synchronized(this){
                logicalclock.incrementAndGet(); // 增加逻辑时钟
                // 修改自己的zxid epoch 
                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
            }
            sendNotifications(); // 发送投票

            // Loop in which we exchange notifications until we find a leader
            while ((self.getPeerState() == ServerState.LOOKING) &&
                    (!stop)){ // 主循环  直到选举出leader
                /*
                 * Remove next notification from queue, times out after 2 times
                 * the termination time
                 */
                //从IO进程里面 获取投票结果，自己的投票也在里面
                Notification n = recvqueue.poll(notTimeout,TimeUnit.MILLISECONDS);

                // 如果没有获取到足够的通知久一直发送自己的选票，也就是持续进行选举
                if(n == null){
                    // 如果空了 就继续发送  直到选举出leader
                    if(manager.haveDelivered()){
                        sendNotifications();
                    } else {
                    // 消息没发出去，可能其他集群没启动 继续尝试连接
                        manager.connectAll();
                    }
                    /// 延长超时时间 
                    int tmpTimeOut = notTimeout*2;
                    notTimeout = (tmpTimeOut < maxNotificationInterval?
                            tmpTimeOut : maxNotificationInterval);
                    LOG.info("Notification time out: " + notTimeout);
                }
                // 收到投票消息 查看是否属于本集群内的消息
                else if(self.getVotingView().containsKey(n.sid)) {
                    switch (n.state) {// 判断收到消息的节点状态
                    case LOOKING:
                        // If notification > current, replace and send messages out
                        // 判断epoch 是否大于 logicalclock ，如是，则是新一轮选举
                        if (n.electionEpoch > logicalclock.get()) {
                            logicalclock.set(n.electionEpoch); // 更新本地logicalclock
                            recvset.clear(); // 清空接受队列
                            // 一次性比较 myid epoch zxid 看此消息是否胜出
                            if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, //此方法看下面代码
                                    getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
                                //投票结束修改票据为 leader票据
                                updateProposal(n.leader, n.zxid, n.peerEpoch);
                            } else {//否则票据不变
                                updateProposal(getInitId(),
                                        getInitLastLoggedZxid(),
                                        getPeerEpoch());
                            }
                            sendNotifications(); // 继续广播票据，让其他节点知道我现在的投票
                         //如果是epoch小于当前  忽略
                        } else if (n.electionEpoch < logicalclock.get()) {
                            break;
                        //如果 epoch 相同 跟上面一样的比较 更新票据 广播票据
                        } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                proposedLeader, proposedZxid, proposedEpoch)) {
                            updateProposal(n.leader, n.zxid, n.peerEpoch);
                            sendNotifications();
                        }

                       // 把最终票据放进接受队列 用来做最后判断
                        recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
                       // 判断选举是否结束 默认算法是否超过半数同意 见下面代码
                        if (termPredicate(recvset,
                                new Vote(proposedLeader, proposedZxid,
                                        logicalclock.get(), proposedEpoch))) {

                            // 一直等待 notification 到达 直到超时就返回null
　　　　　　　　　　　　　　　　 // final static int finalizeWait = 200;
                            while((n = recvqueue.poll(finalizeWait,
                                    TimeUnit.MILLISECONDS)) != null){
                                if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                        proposedLeader, proposedZxid, proposedEpoch)){
                                    recvqueue.put(n);
                                    break;
                                }
                            }
                            // 确定 leader 
                            if (n == null) {
                                // 修改状态
                                self.setPeerState((proposedLeader == self.getId()) ?
                                        ServerState.LEADING: learningState());
                                //返回最终投票结果
                                Vote endVote = new Vote(proposedLeader,
                                                        proposedZxid,
                                                        logicalclock.get(),
                                                        proposedEpoch);
                                leaveInstance(endVote);
                                return endVote;
                            }
                        }
                        break;
                    // 如果收到的选票状态 不是LOOKING 比如刚刚加入已经选举好的集群 
                    // Observer 不参与选举
                    case OBSERVING:
                        LOG.debug("Notification from observer: " + n.sid);
                        break;
                   
                    case FOLLOWING:
                    case LEADING:
                        // 判断 epoch 是否相同
                        if(n.electionEpoch == logicalclock.get()){
                            recvset.put(n.sid, new Vote(n.leader,
                                                          n.zxid,
                                                          n.electionEpoch,
                                                          n.peerEpoch));
                            // 投票是否结束 结束的话确认leader 是否有效
                            // 如果结束 修改自己的投票并且返回
                            if(ooePredicate(recvset, outofelection, n)) {
                                self.setPeerState((n.leader == self.getId()) ?
                                        ServerState.LEADING: learningState());

                                Vote endVote = new Vote(n.leader, 
                                        n.zxid, 
                                        n.electionEpoch, 
                                        n.peerEpoch);
                                leaveInstance(endVote);
                                return endVote;
                            }
                        }

                        //在加入一个已建立的集群之前，确认大多数人都在跟随同一个Leader。
                        outofelection.put(n.sid, new Vote(n.version,
                                                            n.leader,
                                                            n.zxid,
                                                            n.electionEpoch,
                                                            n.peerEpoch,
                                                            n.state));
           
                        if(ooePredicate(outofelection, outofelection, n)) {
                            synchronized(this){
                                logicalclock.set(n.electionEpoch);
                                self.setPeerState((n.leader == self.getId()) ?
                                        ServerState.LEADING: learningState());
                            }
                            Vote endVote = new Vote(n.leader,
                                                    n.zxid,
                                                    n.electionEpoch,
                                                    n.peerEpoch);
                            leaveInstance(endVote);
                            return endVote;
                        }
                        break;
                    default:
                        LOG.warn("Notification state unrecognized: {} (n.state), {} (n.sid)",
                                n.state, n.sid);
                        break;
                    }
                } else {
                    LOG.warn("Ignoring notification from non-cluster member " + n.sid);
                }
            }
            return null;
        } 
　　　// .......
}