ZAB协议
成员发现
成员发现是通过跟随者和领导者交互来完成的，目标是确保大多数节点对领导者的关系没有异议，也就是确立领导者的领导地位。成员发现的实现流程如图所示。
- 1.领导者选举结束，节点进入跟随者状态或者领导者状态后，会分别设置ZAB状态为成员发现状态，具体如下:
1.1 跟随者会调用Follower.followLeader()函数，设置ZAB状态为成员发现状态，如代码所示
self.setZabState(QuorumPeer.ZabState.DISCOVERY);
1.2 领导者会调用Leader.lead()函数，并设置ZAB状态为成员发现状态，如代码所示
self.setZabState(QuorumPeer.ZabState.DISCOVERY);
- 2.跟随者会主动联系领导者，发送自己已接收的领导者任期编号的最大值(也就是acceptedEpoch)的FOLLOWINFO消息给领导者，如代码所示
// 跟领导者建立网络连接
connectToLeader(leaderServer.addr, leaderServer.hostname);
connectionTime = System.currentTimeMills();
// 向领导者报道，并获取领导者的事务标识符最大值
long newEpochZxid = registerWithLeader(Leader.FOLLOWERINFO);
- 3.在接收到来自跟随者的FOLLOWINFO消息后，在LearnerHandler.run()函数中，领导者将创建包含自己的事务标识符最大值的LEADINFO消息，并响应给跟随者，如代码所示
// 创建LEADINFO消息
QuorumPacket newEpochPacket =
new QuorumPacket(Leader.LEADERINFO, newLeaderZxid, ver, null);
// 发送LEADINFO消息给跟随者
oa.writeRecord(newEpochPacket, "packet");
- 4.在接收到来自领导者的LEADINFO消息后，跟随者会基于领导者的任期编号判断领导者是否合法，如果领导者不合法，则发起新的选举，如果领导者合法，则响应ACKEPOCH消息给领导者，如代码所示
// 创建ACKEPOCH消息，包含已提交提案的事务标识符最大值
QuorumPakcet ackNewEpoch =
new QuorumPacket(Leader.ACKEPOCH, lastLoggedZxid, epochBytes, null);
// 响应ACKEPOCH消息给领导者
writePacket(ackNewEpoch, true);
- 5.跟随者设置ZAB状态为数据同步状态，如代码所示
self.setZabState(QuorumPeer.ZabState.SYNCHRONIZATION);
- 6.在LearnerHandler.run()函数中(以及Leader.lead()函数)，领导者会调用waitForEpochAck()函数来阻塞和等待来自大多数节点的ACKEPOCH消息，如代码所示
ss = new StateSummary(bbepoch.getInt(), ackEpochPacket.getZxid());
learnerMaster.waritForEpochAck(this.getSid(), ss)
- 7.在接收到来自大多数节点的ACKEPOCH消息后，在Leader.lead()函数中，领导者设置ZAB状态为数据同步状态。
self.setZabState(QuorumPeer.ZabState.SYNCHRONIZATION);
这样，ZooKeeper就实现了成员发现，且各节点就领导者的领导关系达成了共识。当跟随者和领导者设置ZAB状态为数据同步状态后，它们就进入了数据同步阶段。那么ZooKeeper中的数据同步是如何实现的呢？
数据同步
数据同步也是通过跟随者和领导者交互来完成的。目标是确保跟随者节点上的数据与领导者节点上的数据一直。数据同步的实现流程如图所示。
- 1.在LearnerHandler.run()函数中，领导者调用syncFollower()函数，根据跟随者的事务标识符的最大值判断用哪种方式处理不一致数据，并把已提交提案和未提交提案都同步给跟随者，如代码所示
peerLastZxid = ss.getLastZxid();
boolean needSnap = syncFollower(peerLastZxid, learnerMaster);
在这里，你需要了解领导者向跟随者同步数据的3种方式(TRUNC、DIFF、SNAP)，它们分别代表什么含义呢？要想了解这部分内容，首先要了解一下syncFollower()中3个关键变量的含义。
1.peerLastZxid:跟随者节点上提案的事务标识符欸度最大值
2.maxCommittedLog、minCommittedLog:领导者节点内存队列中已提交提案的事务标识符的最大值和最小值。需要注意的是，maxCommittedLog、minCommittedLog与ZooKeeper的设计有关。在ZooKeeper中，为了更高效地将提案复制到跟随者，领导者会将一定数量(默认值为500)的已提交提案放在内存队列里，而maxCommittedLog、minCommittedLog分别标识的是内存队列中已提交提案的事务标识符最大值和最小值。
说完3个关键变量，再来说说3种同步方式。
1.TRUNC:当peerLastZxid大于maxCommittedLog时，领导者会通知跟随者丢弃超出的那部分提案。比如，如果跟随者的peerLastZxid为11，领导者的maxCommittedLog为10，那么领导者将通知跟随者丢弃事务标识符值为11的提案
2.DIFF:当peerLastZxid小于maxCommittedLog但大于minCommittedLog时，领导者会向跟随者同步缺失的已提交的提案，比如，如果跟随者的peerLastZxid为9，领导者的maxCommittedLog为10，minCommittedLog为9，那么领导者将同步事务标识符值为10的提案给跟随者
3.SNAP:当peerLastZxid小于minCommittedLog时，也就是说，跟随者缺失的提案比较多，那么领导者会同步快照数据给跟随者，并直接覆盖跟随者本地的数据。
在这里，补充一下，领导者先就已提交提案和跟随者达成一致，然后调用learnerMaster.startForwarding()将未提交提案(如果有的话)也缓存发送队列(queuedPackets)，并最终复制给跟随者。也就是说，领导者是以自己的数据为准，实现各节点数据副本的一致的。
需要注意的是，在syncFollower()种，领导者只是将需要发送的差异数据缓存在发送队列，还没有实际发送
- 2.在LearnerHandler.run()函数种，领导者创建NEWLEADER消息并缓存在发送队列种，如代码所示:
// 创建NEWLEADER消息
QuorumPacket newLeaderQP =
new QuorumPacket(Leader.NEWLEADER, newLeaderZxid,learnerMaster.getQuorumVerifierBytes(), null);
// 缓存NEWLEADER消息到发送队列中
queuedPackets.add(newLeaderQP);
- 3.在LearnerHandler.run()函数中，领导者调用startSendingPackets()函数启动一个新线程，并将缓存的数据发送给跟随者，如代码所示
// 发送缓存队列中的数据
startSendingPackets();
- 4.跟随者调用syncWithLeader()函数，处理来自领导者的数据同步，如代码所示
// 处理数据同步
syncWithLEader(newEpochZxid);
- 5.在syncWithLeader()函数中，跟随者在接收到来自领导者的NEWLEADER消息后，返回确认响应给领导者，如代码所示
writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);
- 6.在LearnerHandler.run()函数(以及Leader.lead()函数)中，领导者等待来自大多数节点的NEWLEADER消息的响应，如代码所示
learnerMaster.waitForNewLeaderAck(getSid(), qp.getZxid());
- 7.当接收到来自大多数节点的NEWLEADER消息的响应时，在Leader.lead()函数中，领导者设置ZAB状态为广播状态，如代码所示
self.setZabState(QuorumPeer.ZabState.BROADCAST);
同时，在LearnerHandler.run()中发送UPTODATE消息给所有跟随者，通知它们数据同步已经完成了，如代码所示
queuedPackets.add(new QuorumPacket(Leader.UPTODATE, -1, null, null));
- 8.跟随者在接收到UPTODATE消息后会直到数据不一致已修复，可以处理写请求了，同时设置ZAB状态为广播状态
// 数据同步完成后，跟随者就可以正常处理来自领导者的广播消息了，同时设置ZAB状态为广播状态
self.setZabState(QuorumPeer.ZabState.BROADCAST);
