Zookeeper-leader初始化
在选举完成后，集群每个节点的角色状态就会确定，回到QuorumPeer#start中，每个节点会根据自身的状态完成相应的处理，如下：
- LEADING：创建Leader，等待follower 跟随自身
- FOLLOWING：创建follower，跟随leader并从leader同步数据
- OBSERVING：创建OBSERVING，跟随leader节点
本节我们着重分析leader节点相关的逻辑。
Leader 相关的类
- Leader：leader的控制逻辑类
- LeaderZookeeperServer：leader 处理请求的类
Leader节点初始化，主要是根据节点配置初始化上面两个类，在leader 初始化完成之后会新建用于leader-follower之间通信的连接（接受来自follower的连接）。
Leader 控制逻辑
当leader初始化完成之后通过leader#lead实现leader节点的功能，方法出现异常时，将重置当前节点的状态（LOOKING），进入下一轮选举，下面看下lead的执行流程。
1、载入数据
首先从zkDb中恢复数据（在启动阶段已经载入数据，理论上这里不会再次载入），解析出zxid
- zkDb中恢复数据（启动阶段已经读入数据，理论上这里不会再次读入）
- 解析zxid
- clean up dead session
- 做 snapshot
2、启动 LearnerCnxAcceptor
LearnerCnxAcceptor 主要作用是接受来自follower的请求，并为每一个follower 连接新建一个LearnerHandler 用于处理、同步follower，可参考6集群数据同步。
3、leader 获取选举周期
这一步稍微有点绕，通过Leader#getEpochToPropos实现，步骤如下：
- leader 会调用 Leader#getEpochToPropose 将自身加入到connectingFollowers 这个列表，在没有收到大多数的follower连接上之前，它进入等待状态
- 当一个follower连上leader之后，通过LearnerHandler 中调用Leader#getEpochToPropose将follower加入connectingFollowers 列表，并检查是否大多数follower已经连上leader，如大多数已经连上，则唤醒所有在这个节点等待的follower 和leader，否则它自己也在这里进入等待状态
- 等待的节点会定时去查看大多数follower都连接到leader
实现核心代码如下，可以看出这个通过wait、notify 实现的等待唤醒策略：
long start = Time.currentElapsedTime();
long cur = start;
long end = start + self.getInitLimit()*self.getTickTime();
while(waitingForNewEpoch && cur < end) {
connectingFollowers.wait(end - cur);
cur = Time.currentElapsedTime();
}
等待大部分follower都连接到leader之后，根据每个follower发送epoch 计算出一个新的epoch（选出最大的一个epoch并自增）。leader 会根据这个选举周期初始化这一轮zxid。
4、发送 LEADERINFO 数据包给Follower
当大部分的follower连接上leader后，阻塞等待的follower会被唤醒，接下来leader会发送 LEADERINFO（包括newZxid、version等信息）数据包给follower，然后等待大部分 follower 回复 ack 消息。等待策略和上面差不多，通过Leader#waitForEpochAck 实现。
注：如果follower 的版本小于 0x10000 那么就不会发送LEADERINFO 消息给follower，直接阻塞等待follower 回复的ack消息。在ack消息中也包含follower最新的zxid。现在follower除此回复的消息都是0x10000
这一步主要是将上一步选出来的epoch 广播给集群中的其他节点，其他节点后续会根据这个epoch判断收到的数据包是不是新的leader发送的，避免旧的leader复活之后广播之前的提议而造成状态不一致的问题。
5、Leader通过LearnerHandler异步同步数据给Follower
leader 在收到大多数follower回复ack消息之后，开始和follower 同步数据，在集群对外服务之前确保各个节点中的数据一致。通过LearnerHandler#syncFollower数据同步，同步方式有四种：全量同步（snap）、差异化（diff）、回滚（trunc）、差异化+回滚（diff+trunc） ，具体采用哪种方式还要看follower 和 leader 之间的数据差异情况。
peerLastZxid：该Learner最后处理的ZXID
minCommittedLog：LeadercommittedLog中的最小ZXID
maxCommittedLog：LeadercommittedLog中的最大ZXID
lastProcessedZxid：leader中最后处理的ZXID
强制发送snapshot(测试目的)
-
follower 和 leader 已经同步，则发送空的 diff 消息
if (lastProcessedZxid == peerLastZxid) { // 如果 follower 的peerLastZxid 和 leader lastProcessedZxid相等，说明两个节点的数据一致，不需要进行同步，这时只需要给follower 发送一个diff的包 queueOpPacket(Leader.DIFF, peerLastZxid); needOpPacket = false; needSnap = false; } -
follower 的 txn 比leader 要多，那么则发送 TRUNC ，回滚follower多余的 txn数据。
if (peerLastZxid > maxCommittedLog && !isPeerNewEpochZxid) { // follower的数据比leader 超前，则回滚follower的数据到leader#maxCommittedLog queueOpPacket(Leader.TRUNC, maxCommittedLog); currentZxid = maxCommittedLog; needOpPacket = false; needSnap = false; } -
follower 在committedLog同步的范围内，那么根据follower 的zxid来决定发送 TRUNC 还是DIFF,如果follower 在同步中的话就发送空的DIFF
if ((maxCommittedLog >= peerLastZxid) && (minCommittedLog <= peerLastZxid)) { // follower 数据在处于leader 的 最大最小事务之间，则增量同步 Iterator<Proposal> itr = db.getCommittedLog().iterator(); // 这里有两种情形： // 1、发送 DIFF 给 follower : 对应 follower zxid 在leader的history中，这时只需要将follower中缺失的数据（propose+commit）发过去，然后让follower重做这些提案即可 // 2、发送 TRUN + DIFF 给 follower ： 对应 follower zxid 满足上述条件，但是follower中存在leader中不存在的数据，那么会先发送回滚消息，再增量同步。这种场景在leader 收到propose消息之后还没来得及广播给其他follower就挂掉，新选出的leader没有这个消息，所以需要将此消息回滚 currentZxid = queueCommittedProposals(itr, peerLastZxid, null, maxCommittedLog); needSnap = false; } -
follower 和leader之间的数据差异巨大，follower的最大事务id小于leader 的minCommittedLog。会将leader 磁盘上的txnLog和committedLog同步给follower，如果失败了，会发送snapshot
txnLogSyncEnabled ：是否开启事务日志同步 if (peerLastZxid < minCommittedLog && txnLogSyncEnabled) { // follower 最大的zxid 小于 leader 最小 minCommittedLog，并且允许从txnLog中同步数据 // 计算事务日志允许同步的大小 long sizeLimit = db.calculateTxnLogSizeLimit(); // 如果follower的zxid 在 leader的事务日志中，那么只需要同步事务日志中差异的部分，不需要同步整个snapshot，否则就需要同步snapshot Iterator<Proposal> txnLogItr = db.getProposalsFromTxnLog( peerLastZxid, sizeLimit); if (txnLogItr.hasNext()) { LOG.info("Use txnlog and committedLog for peer sid: " + getSid()); currentZxid = queueCommittedProposals(txnLogItr, peerLastZxid, minCommittedLog, maxCommittedLog); Iterator<Proposal> committedLogItr = db.getCommittedLog().iterator(); currentZxid = queueCommittedProposals(committedLogItr, currentZxid, null, maxCommittedLog); needSnap = false; } // closing the resources if (txnLogItr instanceof TxnLogProposalIterator) { TxnLogProposalIterator txnProposalItr = (TxnLogProposalIterator) txnLogItr; txnProposalItr.close(); } }注：如果需要同步整个镜像，同步过程就不会通过发送每个propose+commit数据给follower，而是leader将它自身的snapshot文件（SNAP）通过网络直接发给follower，follower收到后直接将本地数据覆盖即可。
经过上面的步骤，leader 和follower的差异数据同步完成之后，再次检查leader 和follower 在同步数据期间是否有其他的差异数据，所以最后一步就是同步这部分差异数据。
// 对这个方法的理解是，follower在故障恢复时，如果和leader的数据同步完成，还要再次检查同步数据这段时间是不是leader有了新的提交，需要再次同步
leaderLastZxid = leader.startForwarding(this, currentZxid);
在leader 和follower数据同步完成之后，leader 通过LearnerHandler发送 Leader.NEWLEADER 数据包给所有follower，然后leader通过 leader#waitForNewLeaderAck 阻塞等待大多数follower的ack数据。
6、启动LeaderZooKeeperServer
在大多数的follower都回复了newLeaderAck 数据包之后，这是的leader就成为真正的leader了，它启动LeaderZooKeeperServer ，履行leader 的职责，如下：
- 创建session追踪器（SessionTrackerImpl）
- 启动SessionTracker（它的主要作用是定期清理过期的session）
- 设置请求处理器（构造Leader的请求处理链）
- 注册JMX，然后将当前的服务的状态设置为RUNNING（运行）
所有的LearnerHandler 会阻塞等待LeaderZooKeeperServer 启动完成，之后它开始给所有follower发送Leader.UPTODATE 数据包，之后便处理来自follower数据。它可以收到如下四类消息：
-
Leader.ACK：follower在同步 proposal 之后发给leader的消息；leader收到此消息后直接通过Leader#processAck 处理这条消息。
1、校验消息
if (outstandingProposals.size() == 0) { return; } if (lastCommitted >= zxid) { // The proposal has already been committed return; } Proposal p = outstandingProposals.get(zxid); if (p == null) { return; }2、将收到的消息加入到ack桶中
3、尝试提交消息（通过Leader#tryToCommit，如果没有收到大多数follower的回复时，提交失败）
注：通过tryToCommit可知，在在提交消息时通过在outstandingProposals 中找当前消息的前一条消息，如果找到，说明上一条消息还没提交，则先不提交当前消息。否则提交当前消息，并且通知所有的Observer，最后通过将当前的请求转发给CommitProcessor 继续处理
Leader.PING：leader 和 follower之间的心跳数据。直接通过sessionTracker 更新当前session及超时时间
Leader.REVALIDATE：session校验是否有效
Leader.REQUEST：follower将客户端读以外的操作转发给leader，这一步Leader会通过LeaderZooKeeperServer#submitLearnerRequest将请求首先转发给PrepRequestProcessor。
7、和Follower建立心跳
在Leader 启动完成之后，leader变周期性的发送心跳数据(LearnerHandler#ping)给follower。如果出现follower 掉线或宕机就会将此follower从 learners 中移出，接下来就不会发送propose或者心跳数据给此follower。
当leader上没有大多数的follower时，会将自己关闭，状态重置为LOOKING，进行下一轮的选举。
1、超时时间判断
public synchronized boolean check(long time) { if (currentTime == 0) { return true; } else { long msDelay = (time - currentTime) / 1000000; return (msDelay < (leader.self.tickTime * leader.self.syncLimit)); } } leader.self.tickTime * leader.self.syncLimit:超时时间
至此，Leader的启动流程就已经完成了。
