Zookeeper的领导者选举机制解析

什么是领导者选举

Zookeeper作为一个分布式中间件，为了提高自身的可用性，其内部是多节点以集群模式部署的，官网上的架构图放在下方

根据官网的介绍，每一个Server节点都保存着其他节点的信息，即节点间保持信息互通；只有当集群中大多数节点可用时，整个集群才可用。

集群中的节点有以下几个状态：

LEADER
领导者，集群中有且仅存在一个；其负责集群中其他节点的探活，以及不停地同步最新数据给其他节点
FOLLOWING
跟随者，就是集群中负责工作的小弟
OBSERVER
观察者，这个节点作用和FOLLOWING相似，但不会进行领导者竞选
LOOKING
处于这个状态的节点，说明需要通过领导者选举来确定自己的状态，因此会发送选举信息给其他节点

通过Zookeeper对节点状态的定义可见，领导者选举的目的就是在集群中选出一个主要的节点，其他节点的数据更新都会先发给领导者，再由领导者同步给其他节点。

明白了领导者选举是什么，接着来看领导者选举的过程，其过程的思路在分布式系统设计方面，很值得学习。

领导者选举图解

在整个选举过程中有几个概念要先说明下：

每一个LOOKING节点都是候选人
选举是根据节点的能力，在众多节点中进行多次pk出来的
每次pk的结果叫做选票，会广播给集群中所有候选人

整个领导者选举的过程大致如下，该图示表达了在整个过程中，节点通过选举算法选出一个候选人后，会将该选票通知到其他所有节点。

接着，让我们跟着源码，来看看选举过程的具体实现方式吧。

领导者选举源码解读

既然领导者的选举首先是发生在节点启动时，那么我们就把节点启动作为入口，顺着往下寻找选举的代码实现。

节点启动

集群模式的启动方法是在QuorumPeerMain类中（有集群就有单节点，单节点的启动类是ZooKeeperServerMain），有一个main()，这里省去了无关代码，我们简单来看下；

public class QuorumPeerMain {public static void main(String[] args) {QuorumPeerMain main = new QuorumPeerMain();// 完成初始化配置并启动节点main.initializeAndRun(args);}protected void initializeAndRun(String[] args) throws ConfigException, IOException, AdminServerException {QuorumPeerConfig config = new QuorumPeerConfig();if (args.length == 1) {// 解析配置文件config.parse(args[0]);}if (args.length == 1 && config.isDistributed()) {// 集群模式启动runFromConfig(config);} else {// 单机模式启动ZooKeeperServerMain.main(args);}}
}

还记得Zookeeper的启动参数吧，通常是zkServer.sh start后面会跟上一个config文件路径对吧，这个路径就作为main()的arg[0]参数被传递了进来，具体可以查看zkServer.sh这个脚本。

接着集群模式的启动进入了runFromConfig(config)，这个方法比较长，主要做了这么几件事：

创建一个链接工厂ServerCnxnFactory，这个工厂有两种实现，分别是JDK的NIO和Netty，默认是NIO
接着初始化了一个QuorumPeer实例，并进行一长串的参数赋值
调用quorumPeer.start()启动节点
调用quorumPeer.join()，使main()一直阻塞

如此以来就完成了节点的启动，从以上几个步骤，我们可以得出几个很重要的结论：

QuorumPeer就是代表当前节点的对象实例
QuorumPeer是一个线程实现类
一定实现了线程的run()，并且领导者选举的核心方法就在里面

领导者选举

我们顺着来到QuorumPeer的run()实现，这里主要关注这么一个循环方法，简化如下

@Override
public void run() {while(running){// 枚举当前节点状态switch(getPeerState()){case LOOKING: doSomething...    // 状态为待选举case OBSERVING: doSomething... // 状态为观察者case FOLLOWING: doSomething... // 状态为从节点case LEADING: doSomething...   // 状态为领导节点}}
}

也就是说，当前节点启动之后会进入这个循环，getPeerState()会获取当前节点的状态，接着执行不同的case逻辑，只有当节点是LOOKING状态时才需要领导者选举，于是我们来看LOOKING执行的关键逻辑。

case LOOKING:// shuttingDownLE是个bool标志位，LE是 Leader Election的缩写if (shuttingDownLE) {// 停止领导者选举标志置为falseshuttingDownLE = false;// 领导者选举开始前的准备工作// 1.生成一张投给自己的选票// 2.初始化选举算法，目前仅支持FastLeaderElection，即为默认startLeaderElection();}// 执行选举逻辑，然后保存结果选票setCurrentVote(makeLEStrategy().lookForLeader());break;

看到startLeaderElection()方法名，可能会认为这就是执行选举的方法，然而并不是，这个方法只是在选举前的初始化，注释我写在了代码中，很容易理解。

public synchronized void startLeaderElection() {if (getPeerState() == ServerState.LOOKING) {// 生成一张投给自己的选票currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());}// 初始化选举算法，目前仅支持FastLeaderElection，即为默认this.electionAlg = createElectionAlgorithm(electionType);
}

真正执行选举逻辑的方法是lookForLeader()，尽管这个方法是一个名为Election的接口提供的，但是这个接口只有一个实现类，就是FastLeaderElection，我们直接进入lookForLeader()看选举的核心逻辑

lookForLeader()

在这个方法上有一句注释，非常清楚地解释了这个方法的作用

/*** Starts a new round of leader election. Whenever our QuorumPeer* changes its state to LOOKING, this method is invoked, and it* sends notifications to all other peers.*/

开始新的一轮领导选举，只要当节点状态变成LOOKING时，就会调用这个方法，然后它会将选票发送给其他所有节点。

方法开始首先会给自己投一票，并将选票发送给其他所有节点，调用的是sendNotifications()，注意的是，这个方法并不是调用网络请求直接发送，而是塞进了待发送队列，这个队列就是LinkedBlockingQueue，与之相伴的还有一个接收队列

LinkedBlockingQueue<ToSend> sendqueue;
LinkedBlockingQueue<Notification> recvqueue;

所以，而具体的发送是由一个发送线程去执行的，这个一会再聊。

接着该方法进入一个while循环，第一件事就是从接收队列（recvqueue）中取出一个选票

/*1. Remove next notification from queue, times out after 2 times2. the termination time*/
Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);

我们来看下Notification对象的定义和结构，类上的注释解释了这个类的作用，大概意思表明，这个类是用来在发生选票时，通知其他节点的数据结构。

/**3. Notifications are messages that let other peers know that4. a given peer has changed its vote, either because it has5. joined leader election or because it learned of another6. peer with higher zxid or same zxid and higher server id*/
public static class Notification {/** 格式化版本*/public static final int CURRENTVERSION = 0x2;int version;/** 意向领导*/ long leader;/** 意向领导的zxid，代表数据版本的id*/ long zxid;/** 当前选举的届数*/ long electionEpoch;/** 发送节点的状态*/ QuorumPeer.ServerState state;/** 节点id*/ long sid;QuorumVerifier qv;/** 意向领导处于的届数*/ long peerEpoch;
}

当节点收到一个选票时，会首先判断当前选举的届数，然后再开始选票pk，代码逻辑如下

// 当选票的届数大于当前节点所处的届数，
if (n.electionEpoch > logicalclock.get()) {// 届数更新为最新一届logicalclock.set(n.electionEpoch);// 清空投票箱recvset.clear();// 进行选票pkif (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()) {LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x{}, logicalclock=0x{}",Long.toHexString(n.electionEpoch),Long.toHexString(logicalclock.get()));break;// 当选票的届数等于当前节点所处的届数，直接进行选票pk
} else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {updateProposal(n.leader, n.zxid, n.peerEpoch);sendNotifications();
}

通过注释应该能轻松看懂逻辑，关键的选票pk逻辑就在totalOrderPredicate()中，进入看看

protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {/** We return true if one of the following three cases hold:* 1- New epoch is higher* 2- New epoch is the same as current epoch, but new zxid is higher* 3- New epoch is the same as current epoch, new zxid is the same*  as current zxid, but server id is higher.*/return ((newEpoch > curEpoch)|| ((newEpoch == curEpoch)&& ((newZxid > curZxid)|| ((newZxid == curZxid)&& (newId > curId)))));
}

为了严谨对待选举，这里比较了很多信息，官方也写了注释，我们逐一来看：

先比较选举的届数，届数高的说明是最新一届，胜出
再比较zxid，也就是看谁的数据最新，最新的胜出
最后比较serverid，这是配置文件指定的，节点id大者胜出

通过这个方法就比较出了一个获胜方，接着把这个胜出者封装成Vode对象，放进投票箱中，然后调用sendNotifications()把选票发送到别的节点

// 投票箱是个map，是在刚进入lookForLeader()就实例化了，这里只是为了说明其数据结构
Map<Long, Vote> recvset = new HashMap<Long, Vote>();
// 封装成票，放进投票箱
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

接着会调用voteSet.hasAllQuorums()检查是否收到了所有节点发来的选票，如果还没收齐，就又回到while循环，调用recvqueue.poll()取出选票；如果收齐了，则根据之前的选票pk，已经得出了最终的领导者，若是自己就把自己的状态设为LEADER，否则设为FOLLOWING或者OBSERVING。

至此，集群中的领导者就被选出来了。

总结

我们来回顾一下，在这个过程中，每一个节点都会不断地发送和接收选票，随着优秀节点的选票越来越多，每个节点都在修正自己的判断，最终会选出一个领导者，与所有节点保持信息一致。