zookeeper 选举机制
为什么要进行Leader选举？
Leader主要作用是保证分布式数据一致性，即每个节点的存储的数据同步。遇到以下两种情况需要进行Leader选举
- 服务器初始化启动
- 服务器运行期间无法和Leader保持连接，Leader节点崩溃，逻辑时钟崩溃。
zookeeper提供了三种选举算法,默认的算法是
FastLeaderElection,简单点就是投票数大于半数则胜出：
- LeaderElection
- AuthFastLeaderElection
- FastLeaderElection (default)
概念
myid (服务器ID)
比如有三台服务器，编号分别是1，2，3。编号越大在选举算法中的权重越大。
state (当前服务器的状态)
LOOKING: 竞选状态。FOLLOWING: 随从状态，同步leader状态，参与投票。OBSERVING: 观察状态，同步leader状态，不参与投票。LEADING: 领导者状态。zxid (事务ID)
被推举的Leader事务ID，服务器中存放的最新数据version，值越大说明数据越新，在选举中数据越新权重越大。
electionEpoch (逻辑时钟)
逻辑时钟，用来判断多个投票是否在同一轮选举周期中，该值在服务端是一个自增序列，每次进入新一轮的投票后，都会对该值进行加1操作。
peerEpoch
被推举的Leader的epoch。
选举机制
投票下处理规则
- 首先对比zxid。zxid大的服务器优先作为Leader
- 若zxid相同，比如初始化的时候，每个Server的zxid都为0，就会比较myid，myid大的选出来做Leader。
服务器初始化时选举
目前有3台服务器，每台服务器均没有数据，它们的编号分别是1,2,3按编号依次启动，它们的选择举过程如下：
Server1启动，给自己投票（1,0），然后发投票信息，由于其它机器还没有启动所以它收不到反馈信息，Server1的状态一直属于Looking。
Server2启动，给自己投票（2,0），同时与之前启动的Server1交换结果，由于Server2的编号大所以Server2胜出，但此时投票数正好大于半数，所以Server2成为领导者，Server1成为小弟。
Server3启动，给自己投票（3,0），同时与之前启动的Server1,Server2换信息，尽管Server3的编号大，但之前Server2已经胜出，所以Server3只能成为小弟。
当确定了Leader之后，每个Server更新自己的状态，Leader将状态更新为Leading，Follower将状态更新为Following。
服务器运行期间的选举
zookeeper运行期间，如果有新的Server加入，或者非Leader的Server宕机，那么Leader将会同步数据到新Server或者寻找其他备用Server替代宕机的Server。若Leader宕机，此时集群暂停对外服务，开始在内部选举新的Leader。假设当前集群中有Server1、Server2、Server3三台服务器，Server2为当前集群的Leader，由于意外情况，Server2宕机了，便开始进入选举状态。过程如下
变更状态。其他的非Observer服务器将自己的状态改变为Looking，开始进入Leader选举。
每个Server发出一个投票（myid，zxid），由于此集群已经运行过，所以每个Server上的zxid可能不同。假设Server1的zxid为100，Server3的为99，第一轮投票中，Server1和Server3都投自己，票分别为（1，100）,（3,99）,将自己的票发送给集群中所有机器。
每个Server接收接收来自其他Server的投票，接下来的步骤与启动时步骤相同。
- 默认算法核心代码
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;
synchronized (this) {
//给自己投票
logicalclock.incrementAndGet();
updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
}
//将投票信息发送给集群中的每个服务器
sendNotifications();
//循环，如果是竞选状态一直到选举出结果
while ((self.getPeerState() == ServerState.LOOKING) && (!stop)) {
Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);
//没有收到投票信息
if (n == null) {
if (manager.haveDelivered()) {
sendNotifications();
} else {
manager.connectAll();
}
//...
}
//收到投票信息
else if (self.getCurrentAndNextConfigVoters().contains(n.sid)) {
switch (n.state) {
case LOOKING:
// 判断投票是否过时，如果过时就清除之前已经接收到的信息
if (n.electionEpoch > logicalclock.get()) {
logicalclock.set(n.electionEpoch);
recvset.clear();
//更新投票信息
if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
updateProposal(n.leader, n.zxid, n.peerEpoch);
} else {
updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
}
//发送投票信息
sendNotifications();
} else if (n.electionEpoch < logicalclock.get()) {
//忽略
break;
} 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))) {
// Verify if there is any change in the proposed leader
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) {
self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING : learningState());
Vote endVote = new Vote(proposedLeader, proposedZxid, proposedEpoch);
leaveInstance(endVote);
return endVote;
}
}
break;
case OBSERVING:
//忽略
break;
case FOLLOWING:
case LEADING:
//如果是同一轮投票
if (n.electionEpoch == logicalclock.get()) {
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
//判断是否投票结束
if (termPredicate(recvset, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state))
&& checkLeader(outofelection, n.leader, n.electionEpoch)) {
self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
leaveInstance(endVote);
return endVote;
}
}
//记录投票已经完成
outofelection.put(n.sid, new Vote(n.leader, IGNOREVALUE, IGNOREVALUE, n.peerEpoch, n.state));
if (termPredicate(outofelection, new Vote(n.leader, IGNOREVALUE, IGNOREVALUE, n.peerEpoch, n.state))
&& checkLeader(outofelection, n.leader, IGNOREVALUE)) {
synchronized (this) {
logicalclock.set(n.electionEpoch);
self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
}
Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
leaveInstance(endVote);
return endVote;
}
break;
default:
//忽略
break;
}
} else {
LOG.warn("Ignoring notification from non-cluster member " + n.sid);
}
}
return null;
} finally {
//...
}
}
