Zookeeper是采用的zab协议进行实现的，而不是完全Paxos实现的。在主备系统架构模式下，采用zab协议来保证集群中所有副本的数据一致性。主系统接受处理所有的事务性请求，然后将数据变更状态以proposal提案的形式同步给所有的副本进程。所以在这个过程中，Leader机器显得格外重要。
Leader选举就是在集群中选举出一个主进程，用来接收处理所有客户端的事务性请求。有个隐式条件就是集群中的服务器大于等于2台才能开始Leader选举。
Leader选举时机：
- 服务器启动时期的Leader选举
- 服务器运行期间的Leader选举
选举流程：
- 发送当前自己机器的选票信息给集群中的其它机器
- 接收集群中其它机器发送过来的选票信息
- 处理接收到的投票信息
- 统计投票信息
- 改变当前服务器的状态
选票PK规则：
- 首先比对zxId
- 再比对sid
Zookeeper中leader选举的实现
zookeeper中的leader选举由FastLeaderElection具体实现。其中有几个重要的类：
- Notification：代表收到的投票信息类
- ToSend：发送给其它服务器的投票信息
- WorkerReceiver和WorkerSender以及Messager
protected class Messenger {
// 选票发送器
WorkerSender ws;
// 选票接收器
WorkerReceiver wr;
}
- recvqueue：收票队列
- sendqueue：发送选票队列
WorkerReceiver和WorkerSender不停地从QuorumCnxManager中获取收到的选票信息，以及向集群中所有其它looking机器发送选票信息。
FastLeaderElection继承自Election，实现了其中的选举leader的方法
public Vote lookForLeader() throws InterruptedException {
try {
//TODO 所有收到的选票集合
Map<Long, Vote> recvset = new HashMap<Long, Vote>();
synchronized (this) {
//TODO 逻辑时钟++
logicalclock.incrementAndGet();
//TODO 更新选票 推选的leaderId、zxId 和 选举周期
updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
}
//TODO 给集群中的其它服务器发送当前服务器的投票信息
sendNotifications();
SyncedLearnerTracker voteSet;
//当前服务器是选举状态
while ((self.getPeerState() == QuorumPeer.ServerState.LOOKING) && (!stop)) {
//TODO 从QuorumCnxManager中获取收到的外部 投票信息
FastLeaderElection.Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);
//TODO 收到的外部投票信息为空
if (n == null) {
//TODO 当前服务器选票信息是否发送完
if (manager.haveDelivered()) {
//TODO 发送完了继续发送当前服务器的选票信息给集群中的其它服务器
sendNotifications();
} else {
//TODO 没有发送完就当前服务器建立和其它服务器的链接信息
manager.connectAll();
}
} else if (validVoter(n.sid) && validVoter(n.leader)) {
//TODO 收到的选票中，投票者和被推选者都是 属于投票集合中
//TODO 查看收到的选票的状态
switch (n.state) {
case LOOKING:
if (getInitLastLoggedZxid() == -1) {
LOG.debug("Ignoring notification as our zxid is -1");
break;
}
if (n.zxid == -1) {
LOG.debug("Ignoring notification from member with -1 zxid {}", n.sid);
break;
}
// If notification > current, replace and send messages out
//TODO 如果收到的投票选举周期大于当前机器的时钟周期
if (n.electionEpoch > logicalclock.get()) {
//TODO 更新当前机器的时钟周期
logicalclock.set(n.electionEpoch);
//TODO 清空所有收到的投票信息
recvset.clear();
//TODO 如果收到的选票信息优于当前服务器选票信息，变更当前服务器的投票信息
if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
updateProposal(n.leader, n.zxid, n.peerEpoch);
} else {
//TODO 更新选票为自己当前服务器的选票信息
updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
}
//TODO 发送选票信息
sendNotifications();
} else if (n.electionEpoch < logicalclock.get()) {
//TODO 如果逻辑时钟小于当前逻辑时钟，忽略
LOG.debug(
"Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x{}, logicalclock=0x{}",
Long.toHexString(n.electionEpoch),
Long.toHexString(logicalclock.get()));
break;
} else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {
//TODO 选举周期相同，但是收到的选票更优，更新选票信息
updateProposal(n.leader, n.zxid, n.peerEpoch);
//TODO 发送选票信息
sendNotifications();
}
//TODO 收到的选票信息放入集合中
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
//TODO 投票归档，查看是否已leader选举完成
voteSet = getVoteTracker(recvset, new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch));
if (voteSet.hasAllQuorums()) {
while ((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {
recvqueue.put(n);
break;
}
}
if (n == null) {
setPeerState(proposedLeader, voteSet);
Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch);
leaveInstance(endVote);
return endVote;
}
}
break;
case OBSERVING:
LOG.debug("Notification from observer: {}", n.sid);
break;
case FOLLOWING:
case LEADING:
if (n.electionEpoch == logicalclock.get()) {
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
voteSet = getVoteTracker(recvset, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
if (voteSet.hasAllQuorums() && checkLeader(recvset, n.leader, n.electionEpoch)) {
setPeerState(n.leader, voteSet);
Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
leaveInstance(endVote);
return endVote;
}
}
outofelection.put(n.sid, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
voteSet = getVoteTracker(outofelection, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
if (voteSet.hasAllQuorums() && checkLeader(outofelection, n.leader, n.electionEpoch)) {
synchronized (this) {
logicalclock.set(n.electionEpoch);
setPeerState(n.leader, voteSet);
}
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 {
if (!validVoter(n.leader)) {
LOG.warn("Ignoring notification for non-cluster member sid {} from sid {}", n.leader, n.sid);
}
if (!validVoter(n.sid)) {
LOG.warn("Ignoring notification for sid {} from non-quorum member sid {}", n.leader, n.sid);
}
}
}
return null;
} finally {
try {
if (self.jmxLeaderElectionBean != null) {
MBeanRegistry.getInstance().unregister(self.jmxLeaderElectionBean);
}
} catch (Exception e) {
LOG.warn("Failed to unregister with JMX", e);
}
self.jmxLeaderElectionBean = null;
LOG.debug("Number of connection processing threads: {}", manager.getConnectionThreadCount());
}
}
