分布式与一致性协议之ZAB协议(六)

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);