面试官问我zookeeper选举过程，我当场给他讲了源码

没有故事的陈师傅

发布于 2021-04-09 15:16:16

3450

发布于 2021-04-09 15:16:16

文章被收录于专栏：运维开发故事运维开发故事

集群概述

zookeper 在生产环境中通常都是通过集群方式来部署的，以保证高可用, 下面是 zookeeper 官网给出的一个集群部署结构图：

从上图可以得出， zookeeper server 的每个节点都和主节点保持通讯的，每个节点上面都存储有数据和日志的备份，只有当大多数节点可用集群才是可用的。本文主要是基于 zookeeper 3.8.0 讲解，主要是通过源码的维度来分析 zookeeper 选举过程对于 zookeeper 的源码编译大家可以参考：编译运行Zookeeper源码

集群节点状态

集群节点状态定义在 QuorumPeer#ServerState 枚举，主要是包含 LOOKING、FOLLOWING、LEADING、OBSERVING 四个状态, 下面是定义的代码和说明

public enum ServerState {
    // 寻找leader状态。当服务器处于该状态时，它会认为当- 前集群中没有leader，因此需要进入leader选举状态。
    LOOKING,
    // 跟随者状态。表明当前服务器角色是 follower。
    FOLLOWING,  
    // 领导者状态。表明当前服务器角色是 leader。
    LEADING,
    // 观察者状态。表明当前服务器角色是 observer。
    OBSERVING
}

Leader 选举过程

启动和初始化

QuorumPeerMain 是 zookeeper 的启动类，通过 main 方法启动

// 不展示非核心代码
public static void main(String[] args) {
    QuorumPeerMain main = new QuorumPeerMain();
    main.initializeAndRun(args);
}
protected void initializeAndRun(String[] args) throws ConfigException, IOException, AdminServerException {
    // 集群模式启动
    if (args.length == 1 && config.isDistributed()) {
        runFromConfig(config);
    } else {
    }
}
public void runFromConfig(QuorumPeerConfig config) throws IOException, AdminServerException {
    // quorumPeer 启动
 quorumPeer.start();
}

QuorumPeer 是一个线程实例类，当调用 start 方法过后会致性 QuorumPeer#run() 方法，进行集群状态的判断最终进入是否执行选举或者同步集群节点数据信息等一系列的操作，下面是核心代码：

@Override
public void run() {
    try {
        while (running) {
            switch (getPeerState()) {
                case LOOKING:
                    // 投票给自己
                    setCurrentVote(makeLEStrategy().lookForLeader());
                    break;
                case OBSERVING:
                    setObserver(makeObserver(logFactory));
                    observer.observeLeader();
                    break;
                case FOLLOWING:
                    setFollower(makeFollower(logFactory));
                    follower.followLeader();
                    break;
                case LEADING:
                    setLeader(makeLeader(logFactory));
                    leader.lead();
                    setLeader(null);
                    break;
            }
        }
    } finally {
    }
}

进行选举

FastLeaderElection 是选举的核心类，在这个类里面有对投票和选票的处理过程

public Vote lookForLeader() throws InterruptedException {
    // 创建一个当前选举周期的投票箱
    Map<Long, Vote> recvset = new HashMap<Long, Vote>();
    // 创建一个投票箱。这个投票箱和recvset 不一样。
    // 存储当前集群中如果已经存在Leader了的投票
    Map<Long, Vote> outofelection = new HashMap<Long, Vote>();
    int notTimeout = minNotificationInterval;
    synchronized (this) {
        // 递增本地选举周期
        logicalclock.incrementAndGet();
        // 为自己投票
        updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
    }
    // 广播投票
    sendNotifications();
    SyncedLearnerTracker voteSet = null;
    // 如果当前服务器的状态为Looking，和stop参数为false，那么进行选举
    while ((self.getPeerState() == ServerState.LOOKING) && (!stop)) {
        if (n.electionEpoch > logicalclock.get()) {
            logicalclock.set(n.electionEpoch);
            recvset.clear();
            // totalOrderPredicate 投票 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 (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {
            updateProposal(n.leader, n.zxid, n.peerEpoch);
            sendNotifications();
        }
        // 监听通信层接收的投票
        Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);
        // 放入投票箱
        recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
        // 过半逻辑
        voteSet = getVoteTracker(recvset, new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch));
    }
}

totalOrderPredicate 主要是选票 PK 的逻辑，我们再来看看代码：

protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
    if (self.getQuorumVerifier().getWeight(newId) == 0) {
        return false;
    }
    return ((newEpoch > curEpoch)
            || ((newEpoch == curEpoch)
                && ((newZxid > curZxid)
                    || ((newZxid == curZxid)
                        && (newId > curId)))));
}

选举过程是这个样子的，其实官方也给出了注释：

先比较选举的届数，届数高的说明是最新一届，胜出
再比较zxid，也就是看谁的数据最新，最新的胜出
最后比较serverid，这是配置文件指定的，节点id大者胜出选举完成后通过 sendNotifications(); 通知其他的节点。

过程总结

前面我粗略的讲解 zookeeper 从启动过程在到选举，选举结果同步的，以及如何进行投票的选举结果确认过程，但是 zookeeper 作为一个高性能、高可靠的分布式协调中间件，在很多设计的细节也是非常的优秀的。

投票过程

通常情况下，在投票的过程中 zxid 越大越有可能成为 leader 主要是由于 zxid 越大该节点的数据越多，这样的话就可以减少数据的同步过程中节点事务的撤销和日志文件同步的比较过程，以提升性能。下面是 5 个 zookeeper 节点选举的过程。

注: (sid, zxid), 当前场景为 server1 ,server2 出现故障， server3 的 zxid = 9 ， server4 和 server5 的 zxid 为 8. 进行两轮选举，最终选出 sever3 为 leader 节点

多层网络架构

在前面的分析过程中我省略了 Zookeeper 节点之间通讯的 NIO 操作，这部分简单来讲 zookeeper 将他们划分为传输层和业务层。通过 SendWorker、RecvWorker 处理网络层数据包， WorkerSender 和 WorkerReceiver 处理业务层的数据。