前面分析这么多，还没有正式分析到 leader 选举的核心流程，前期准备工作做好了以后，接下来就开始正式分析 leader 选举的过程：
public synchronized void start() {
loadDataBase();
cnxnFactory.start();
startLeaderElection();
super.start(); //启动线程
}
很明显，super.start() 表示当前类 QuorumPeer 继承了线程，线程必须要重写 run 方法，所以我们可以在 QuorumPeer 中找到一个 run 方法。
QuorumPeer.run
这段代码的逻辑比较长。粗略看一下结构，好像也不难
PeerState 有几种状态，分别是
- LOOKING，竞选状态。
- FOLLOWING，随从状态，同步 leader 状态，参与投票。
- OBSERVING，观察状态,同步 leader 状态，不参与投票。
- LEADING，领导者状态。
对于选举来说，默认都是 LOOKING 状态，只有 LOOKING 状态才会去执行选举算法。每个服务器在启动时都会选择自己做为领导，然后将投票信息发送出去，循环一直到选举出领导为止。
@Override
public void run() {
setName("QuorumPeer" + "[myid=" + getId() + "]" + cnxnFactory.getLocalAddress());
// … 根据选举状态，选择不同的处理方式
while (running) {
switch (getPeerState()) {
case LOOKING:
LOG.info("LOOKING");
// 判断是否为只读模式,通过”readonlymode.enabled”开 启
if (Boolean.getBoolean("readonlymode.enabled")) {
// 只读模式的启动流程
} else {
try {
setBCVote(null);
// 设置当前的投票，通过策略模式来决定当前用哪个选举算法来进行领导选举
setCurrentVote(makeLEStrategy().lookForLeader());
} catch (Exception e) {
LOG.warn("Unexpected exception", e);
setPeerState(ServerState.LOOKING);
}
}
break;
// …后续逻辑暂时不用管
}
}
}
FastLeaderElection.lookForLeader
开始发起投票流程
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();//更新逻辑时钟，用来判断是否在同一轮选举周期
//初始化选票数据：这里其实就是把当前节点的 myid，zxid，epoch 更新到本地的成员属性
updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
}
LOG.info("New election. My id = " + self.getId() + ", proposed zxid=0x" + Long.toHexString(proposedZxid));
sendNotifications();//异步发送选举信息
/*
* Loop in which we exchange notifications until we find a leader
*/
//这里就是不断循环，根据投票信息进行进行 leader 选举
while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){
/*
* Remove next notification from queue, timesout after 2 times
* the termination time
*/
//从 recvqueue 中获取消息
Notification n = recvqueue.poll(notTimeout,TimeUnit.MILLISECONDS);
/*
* Sends more notifications if haven't received enough.
* Otherwise processes new notification.
*/
if(n == null){ //如果没有获取到外部的投票，有可能是集群之间的节点没有真正连接上
if(manager.haveDelivered()){//判断发送队列是否有数据，如果发送队列为空，再发一次自己的选票
sendNotifications();
} else {//在此发起集群节点之间的连接
manager.connectAll();
}
/*
* Exponential backoff
*/
int tmpTimeOut = notTimeout*2;
notTimeout = (tmpTimeOut < maxNotificationInterval? tmpTimeOut : maxNotificationInterval);
LOG.info("Notification time out: " + notTimeout);
}
}
}
}
选票的判断逻辑（核心代码）
// 判断收到的选票中的 sid 和选举的 leader 的 sid 是否存在于我们集群所配置的 myid 范围
if (validVoter(n.sid) && validVoter(n.leader)) {
// 判断接收到的投票者的状态，默认是 LOOKING 状态,说明当前发起投票的服务器也是在找 leader
switch (n.state) {
case LOOKING: // 说明当前发起投票的服务器也是在找 leader
// 如果收到的投票的逻辑时钟大于当前的节点的逻辑时钟
if (n.electionEpoch > logicalclock.get()) {
logicalclock.set(n.electionEpoch);// 更新成新一轮的逻辑时钟
recvset.clear();
// 比较接收到的投票和当前节点的信息进行比较，比较的顺序epoch、zxid、myid,如果返回
// true，则更新当前节点的票据（sid,zxid,epoch）,那么下次再发起投票的时候，就不再是选自己了
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()) {// 如果小于，说明收到的票据已经过期了，直接把这张票丢掉
if (LOG.isDebugEnabled()) {
LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
+ Long.toHexString(n.electionEpoch) + ", logicalclock=0x"
+ Long.toHexString(logicalclock.get()));
}
break; // 这个判断表示收到的票据的 epoch 是相同的，那么按照 epoch、zxid、myid
// 顺序进行比较比较成功以后，把对方的票据信息更新到自己的节点
} else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid,
proposedEpoch)) {
updateProposal(n.leader, n.zxid, n.peerEpoch);
sendNotifications();// 把收到的票据再发出去告诉大家我要选 n.leader 为 leader
}
if (LOG.isDebugEnabled()) {
LOG.debug("Adding vote: from=" + n.sid + ", proposed leader=" + n.leader + ", proposed zxid=0x"
+ Long.toHexString(n.zxid) + ", proposed election epoch=0x"
+ Long.toHexString(n.electionEpoch));
}
// 将收到的投票信息放入投票的集合 recvset 中, 用来作为最终的 "过半原则" 判断
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
// 判断选举是否结束
if (termPredicate(recvset, new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch))) {
// 进入这个判断，说明选票达到了 leader 选举的要求
// 在更新状态之前，服务器会等待 finalizeWait 毫秒时间来接收新的选票，以防止漏下关键选票如果收到可能改变
// Leader 的新选票，则重新进行计票
while ((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid,
proposedEpoch)) {
recvqueue.put(n);
break;
}
}
// 如果 notifaction 为空，说明 Leader 节点是可以确定好了
if (n == null) {
// 设置当前当前节点的状态（判断 leader 节点是不是我自己，如果是，直接更新当前节点的 state 为
// LEADING）否则，根据当前节点的特性进行判断，决定是FOLLOWING 还是 OBSERVING
self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING : learningState());
// 组装生成这次 Leader 选举最终的投票的结果
Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch);
leaveInstance(endVote);// 清空
return endVote; // 返回最终的票据
}
}
break;
case OBSERVING:// OBSERVING 不参与 leader 选举
LOG.debug("Notification from observer: " + n.sid);
break;
case FOLLOWING:
case LEADING:
/*
* Consider all notifications from the same epoch together.
*/
if (n.electionEpoch == logicalclock.get()) {
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
if (ooePredicate(recvset, outofelection, n)) {
self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
leaveInstance(endVote);
return endVote;
}
}
/*
* Before joining an established ensemble, verify a majority is
* following the same leader.
*/
outofelection.put(n.sid, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
if (ooePredicate(outofelection, outofelection, n)) {
synchronized (this) {
logicalclock.set(n.electionEpoch);
self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
}
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;
}
}
投票处理的流程图
termPredicate
这个方法是使用过半原则来判断选举是否结束，如果返回 true，说明能够选出 leader 服务器votes 表示收到的外部选票的集合vote 表示但前服务器的选票。
protected boolean termPredicate(HashMap<Long, Vote> votes, Vote vote) {
HashSet<Long> set = new HashSet<Long>();
// 遍历接收到的所有选票数据
for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
// 对选票进行归纳，就是把所有选票数据中和当前节点的票据相同的票据进行统计
if (vote.equals(entry.getValue())) {
set.add(entry.getKey());
}
} // 对选票进行判断
return self.getQuorumVerifier().containsQuorum(set);
}
QuorumMaj. containsQuorum
判断当前节点的票数是否是大于一半，默认采用 QuorumMaj 来实现。
public boolean containsQuorum(Set<Long> set){
return (set.size() > half);
}
这个 half 的值是多少呢？
可以在 QuorumPeerConfig.parseProperties 这个方法中，找到如下代码。
也就是说，在构建 QuorumMaj 的时候，传递了当前集群节点的数量，这里是 3那么，hafl=3/2=1
public QuorumMaj(int n){
this.half = n/2;
}
那么 set.size()>1. 意味着至少要有两个节点的票据是选择你当 leader，否则，还得继续投。