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) { // 停止领导者选举标志置为false shuttingDownLE = 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 times 2. the termination time */ Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);

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

/** 3. Notifications are messages that let other peers know that 4. a given peer has changed its vote, either because it has 5. joined leader election or because it learned of another 6. 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(); // 进行选票pk 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()) { 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。

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

总结

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