Zookeeper源码分析：选举流程

参考资料

<<从PAXOS到ZOOKEEPER分布式一致性原理与实践>>
zookeeper-3.0.0

Zookeeper选举模式

针对zookeeper-3.0.0版本，选举模式可以分为三种模式，主要分为快速认证选举，快速选举和直接UDP选举这三个算法。通过选举来确定启动服务的角色是leader还是follower。Zookeeper集群规定至少是两台服务器，并且建议是以奇数组成服务器集群，默认以三台举例，并且服务器myid依次为1、2、3。

集群启动时选举流程

每个Server启动后发出一个投票。因为是在集群初始启动阶段，对于机器而言都是将自己作为Leader服务器进行投票，每次投票包含的数据内容就是：所推举的服务器的myid和zxid，zxid就是以自增的值去生成，例如刚刚开始生成的zxid为0，如果还要继续投票循环则生成为1，此时的投票信息就是(1,0)，（2，0），（3，0）。
接受来自各个服务器的投票。每个服务器都会接受到其他服务器的投票。集群中的每个服务器在接受到投票后，首先会判断该投票的有效性，包括检查是否是本轮投票，是否来自LOOKING状态的服务器等信息的检查，检查合法后才会处理计算投票。
处理投票。在接受到来自其他服务器的投票后，针对每一个投票，服务器都需要将别人的投票和自己的投票进行对比，首先会检查zxid，如果zxid较大则选择zxid较大的服务器作为leader，如果zxid相同，那就比较myid，myid较大的服务器作为leader。由于是集群启动阶段故zxid为0 都相同，此时Server1和Server2会将当前的投票信息更新为(3,0)然后将该选票发送出去。
统计投票。每次投票后，服务器都统计投票，判断是否已经有过半的机器接收到相同的投票信息。此时统计出集群中其他两天机器已经接受了(3,0)这个投票信息，故达到了大于集群数量一半的要求。
改变服务器状态。此时Server3确定角色为leader，Server1和Server2确定角色为follower。

集群运行过程中选举流程

在集群的运行过程中，一旦选出了一个leader，那么所有服务器的集群角色一般不会再发生改变，即leader服务器一直作为集群的leader，即使集群中有非leader集群挂了或是有新机器加入集群也不会影响leader，如果leader服务器挂了，此时整个集群就暂时无法对外提供服务，会进入新一轮的leader选举。此时的选举流程基本保持和集群启动时的流程保持一致，只不过在写入投票阶段过程中生成的zxid会根据运行的状态去确定该内容值。

变更状态。首先当leader挂了之后，剩余的服务器都会将自己的服务器状态变更为looking状态，然后进入选举流程。
每个Server发送一个投票。在这个过程中，需要生成投票信息，因为是运行期间，因此每个zxid可能会不同，如果Server1为10，Server2为11，Server3为12，此时投票就会选择Server3的12然后将投票发送出去。剩余的流程就重复如上。

选举算法分析

在上文集群启动中，最后会执行到QuorumPeer类中的run方法中去，其中就是当处于LOOKING状态时，就进行选举操作。

                     case LOOKING:try {LOG.info("LOOKING");setCurrentVote(makeLEStrategy().lookForLeader());           // 设置投票并选择leader} catch (Exception e) {LOG.warn("Unexpected exception",e);                         // 如果出错则设置为LOOKING状态setPeerState(ServerState.LOOKING);}break;

此时就执行了makeLEStrategy函数返回的类的lookForLeader方法。

    protected Election makeLEStrategy(){if(electionAlg==null)return new LeaderElection(this);return electionAlg;}

在集群模式启动时，默认不选择选举算法时，此时的electionAlg就是null，此时就会新生成一个LeaderElection类实例，并调用该类的lookForLeader方法，在启动模式的时候，会默认启动ResponderThread类中的UDP选举算法，该类就是监听执行的端口，来获取其他服务器的选票信息。并且该类处于选举中的执行流程如下；

                        switch (getPeerState()) {                                   case LOOKING:                                                   // 如果是竞选状态responseBuffer.putLong(current.id);                         // 压入id 和 zxidresponseBuffer.putLong(current.zxid);break;

就是将当前的id和zxid压入返回数据中。此时我们继续查看，LeaderElection的选举方法。

    private ElectionResult countVotes(HashMap<InetSocketAddress, Vote> votes, HashSet<Long> heardFrom) {ElectionResult result = new ElectionResult();                                   // 初始化选举结果实例// Initialize with null voteresult.vote = new Vote(Long.MIN_VALUE, Long.MIN_VALUE);                         // 初始化一个默认的vote result.winner = new Vote(Long.MIN_VALUE, Long.MIN_VALUE);                       // 初始化一个默认的winner Collection<Vote> votesCast = votes.values();                                    // 获取所有的值// First make the views consistent. Sometimes peers will have// different zxids for a server depending on timing.for (Iterator<Vote> i = votesCast.iterator(); i.hasNext();) {                   // 迭代遍历该值Vote v = i.next();if (!heardFrom.contains(v.id)) {                                            // 如果不包括该值则移除// Discard votes for machines that we didn't hear fromi.remove();continue;}for (Vote w : votesCast) {                                                  // 遍历votesCastif (v.id == w.id) {                                                     // 如果id相同if (v.zxid < w.zxid) {                                              // 并且w的zxid大于当前的v的zxidv.zxid = w.zxid;                                                // 将v的zxid更新为w的zxid}}}}HashMap<Vote, Integer> countTable = new HashMap<Vote, Integer>();               // 统计选票hashmap// Now do the tallyfor (Vote v : votesCast) {                                                      // 遍历所有voteInteger count = countTable.get(v);                                          // 首先获取一下该v 如果为NULL则统计数设置为1if (count == null) {count = Integer.valueOf(0);}countTable.put(v, count + 1);                                               // 将当前v压入并计数加1if (v.id == result.vote.id) {                                               // 如果当前压入的v的id 与投票获取的id相同则加1result.count++;} else if (v.zxid > result.vote.zxid|| (v.zxid == result.vote.zxid && v.id > result.vote.id)) {         // 如果当前的zxid大于当前投票获取的vote的zxid或者 当前的zxid与投票获取的zxid相同当时当前id大于投票获取的idresult.vote = v;                                                        // 设置当前的vote为 v  此时就体现了获取最大zxid 或者在zxid相同的情况下 选举最大server的id的选举思路result.count = 1;}}result.winningCount = 0;                                                        // 设置选举统计数为0LOG.warn("Election tally: ");for (Entry<Vote, Integer> entry : countTable.entrySet()) {                      // 遍历所有的列表if (entry.getValue() > result.winningCount) {                               // 如果获取的统计数大于当前结果的选举数则设置统计数与最终获胜的选票result.winningCount = entry.getValue();result.winner = entry.getKey();}LOG.warn(entry.getKey().id + "\t-> " + entry.getValue());}return result;                                                                  // 返回结果}public Vote lookForLeader() throws InterruptedException {self.setCurrentVote(new Vote(self.getId(), self.getLastLoggedZxid()));      // 设置当前选票 设置为自己服务器的id 和自己的zxid// We are going to look for a leader by casting a vote for ourselfbyte requestBytes[] = new byte[4];ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);byte responseBytes[] = new byte[28];ByteBuffer responseBuffer = ByteBuffer.wrap(responseBytes);/* The current vote for the leader. Initially me! */DatagramSocket s = null;try {s = new DatagramSocket();                                               // 设置UDP通信s.setSoTimeout(200);                                                    // 设置发送超时时间} catch (SocketException e1) {e1.printStackTrace();                                                   // 如果报错则退出System.exit(4);}DatagramPacket requestPacket = new DatagramPacket(requestBytes,requestBytes.length);                                               // 设置发送数据包DatagramPacket responsePacket = new DatagramPacket(responseBytes,responseBytes.length);                                              // 设置接收数据包HashMap<InetSocketAddress, Vote> votes = new HashMap<InetSocketAddress, Vote>(self.quorumPeers.size());                                           // 保存投票信息int xid = new Random().nextInt();                                           // 获取xidwhile (self.running) {                                                      // 循环执行votes.clear();requestBuffer.clear();requestBuffer.putInt(xid);                                              // 压入xidrequestPacket.setLength(4);HashSet<Long> heardFrom = new HashSet<Long>();for (QuorumServer server : self.quorumPeers.values()) {                 // 遍历其他服务器列表requestPacket.setSocketAddress(server.addr);                        // 设置发送的端口LOG.warn("Server address: " + server.addr);try {s.send(requestPacket);                                          // 发送数据responsePacket.setLength(responseBytes.length);s.receive(responsePacket);                                      // 接收返回数据if (responsePacket.getLength() != responseBytes.length) {       // 检验返回数据是否合法LOG.error("Got a short response: "+ responsePacket.getLength());continue;}responseBuffer.clear();int recvedXid = responseBuffer.getInt();                        // 获取接收到的xid if (recvedXid != xid) {                                         // 如果不相同则继续LOG.error("Got bad xid: expected " + xid+ " got " + recvedXid);continue;}long peerId = responseBuffer.getLong();                         // 获取返回的id heardFrom.add(peerId);//if(server.id != peerId){Vote vote = new Vote(responseBuffer.getLong(), responseBuffer.getLong());                              // 设置选票的信息InetSocketAddress addr = (InetSocketAddress) responsePacket.getSocketAddress();votes.put(addr, vote);                                      // 压入hash表中统计选票//}} catch (IOException e) {LOG.error("Error in looking for leader", e);// Errors are okay, since hosts may be// down// ZooKeeperServer.logException(e);}}ElectionResult result = countVotes(votes, heardFrom);                   // 统计选票if (result.winner.id >= 0) {                                            // 如果id大于等于0 self.setCurrentVote(result.vote);                                   // 设置当前的选票if (result.winningCount > (self.quorumPeers.size() / 2)) {          // 如果获胜的统计大于总数的一半self.setCurrentVote(result.winner);                             // 设置当前的选票为获胜者s.close();Vote current = self.getCurrentVote();                           // 获取当前选票self.setPeerState((current.id == self.getId()) ? ServerState.LEADING: ServerState.FOLLOWING);          // 如果当前id与当前选票id相同则是主否则为从if (self.getPeerState() == ServerState.FOLLOWING) {             // 如果当前状态为FOLLOWING则休眠后返回当前选票Thread.sleep(100);}return current;}}Thread.sleep(1000);                                                     // 如果没有找出则休眠继续下一个循环}return null;}

从寻找主的过程中可以看出，选举的思路就是选取最大的zxid，如果zxid相同则选择最大myid的Server最为统计的获胜者。

总结

本文主要就是概述了选举算法的一个基本流程与实现的思路，选举的流程实现较为简单，主要就是根据固定的规则，首先选择最大的zxid，如果zxid相同则选择最大的server的id的思路，并且通过UDP来访问各个服务器之间的数据的交互，从而完成选举过程。由于本人才疏学浅，如有错误请批评指正。