Zookeeper启动和集群选举

1. QuorumPeerMain运行

1）判断是采用单实例模式仍是多实例模式启动QuorumPeerMain

2）在多实例模式下，加载启动参数中指定的配置文件

3）启动QuorumPeer

public class QuorumPeerMain {
    ...

    protected QuorumPeer quorumPeer;

    public static void main(String[] args) {
        QuorumPeerMain main = new QuorumPeerMain();
        try {
            main.initializeAndRun(args);
        }
        ...
    }

    protected void initializeAndRun(String[] args) throws ConfigException, IOException {
        QuorumPeerConfig config = new QuorumPeerConfig();
        if (args.length == 1) {
            config.parse(args[0]);
        }
        // 启动data目录下的定时清理任务
        ...
        // 启动参数中指定配置文件，且配置文件中指定为多实例
        if (args.length == 1 && config.servers.size() > 0) {
            runFromConfig(config); //
        } else {
          // 单实例模式启动
          ...
        }
    }

    public void runFromConfig(QuorumPeerConfig config) throws IOException {
      ...
      try {
          // 客户端链接工厂，默认NIOServerCnxnFactory
          ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();           cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns());
          quorumPeer = new QuorumPeer(config.getServers(),
                                      new File(config.getDataDir()),
                                      new File(config.getDataLogDir()),
 config.getElectionAlg(), // 选举算法，默认FastLeaderElection（3）
                                      config.getServerId(),
                                      config.getTickTime(),
                                      config.getInitLimit(),
                                      config.getSyncLimit(),
                                      config.getQuorumListenOnAllIPs(),
                                      cnxnFactory,
                                      config.getQuorumVerifier());
          ...
 quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));           ...
 quorumPeer.start(); // 启动当前实例
          quorumPeer.join(); // 主线程等待quorumPeer线程执行
      } catch (InterruptedException e) {
          ...
      }
    }
}

2. QuorumPeer启动

1）监听客户端链接

2）确认选举算法，监听选举端口

3）启动服务（run），开始选举

4 ) 成为Leader / Follower / Observer（后续介绍）

public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider {
    ...
    ServerCnxnFactory cnxnFactory; // NIOServerCnxnFactory
    ...
    @Override
    public synchronized void start() {
        loadDataBase(); // 加载currentEpoch和acceptedEpoch
        cnxnFactory.start(); // 监听客户端链接，NIOServerCnxnFactory.start
        startLeaderElection(); // 确认选举算法，监听选举端口
        super.start(); // 启动服务（run），开始选举
    }
    ...
    synchronized public void startLeaderElection() {
        try {
            // 当前投票投给本身
            currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
        } catch(IOException e) {
            ...
        }
        ...
        // 确认选举算法，监听选举端口
        this.electionAlg = createElectionAlgorithm(electionType);
    }
    ...
    protected Election createElectionAlgorithm(int electionAlgorithm){
        Election le=null;
        switch (electionAlgorithm) {
        ...
        case 3:
            qcm = createCnxnManager(); // QuorumCnxManager负责收发投票
            QuorumCnxManager.Listener listener = qcm.listener;
            if(listener != null){
                listener.start(); // 启动ServerSocket监听选举端口
                le = new FastLeaderElection(this, qcm);             }
            ...
            break;
        ...
        }
        return le;
    }
    ...
    @Override
    public void run() {
        ...
        try {
            while (running) {
                switch (getPeerState()) {
                case LOOKING:
                    if (Boolean.getBoolean("readonlymode.enabled")) {
                        // 启动ReadOnlyZooKeeperServer处理只读请求
                        ...
                    } else {
                        try {
                            setBCVote(null);
                            // 开始选举，FastLeaderElection.lookForLeader
 setCurrentVote(makeLEStrategy().lookForLeader());                         }
                        ...
                    }
                    break;
                case OBSERVING:
                    try {
 setObserver(makeObserver(logFactory)); // Observer observer.observeLeader();                     }
                    ...
                    break;
                case FOLLOWING:
                    try {
 setFollower(makeFollower(logFactory)); // Follower follower.followLeader();                     }
                    ...
                    break;
                case LEADING:
                    try {
 setLeader(makeLeader(logFactory)); // Leader leader.lead();                         setLeader(null);
                    }
                    ...
                    break;
                }
            }
        } finally {
            ...
        }
    }
}

 

3. 快速选举算法

1）广播推荐的Leader为本身，并接收其它ZK服务器的投票

2）选举进行中

　　1' 若对方推荐的Leader > 本身推荐Leader，跟随对方投票并广播

　　2' 更新票箱，判断对方投票是否过半，是则返回最终投票，不然选举继续

3）选举已经结束

　　更新票箱，判断对方投票是否过半且为Leader，是则返回最终投票，不然选举继续

public class FastLeaderElection implements Election {
    ...
    public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager){
        ...
        starter(self, manager);
    }
    private void starter(QuorumPeer self, QuorumCnxManager manager) {
        ...
        sendqueue = new LinkedBlockingQueue<ToSend>(); // 待发送投票队列
        recvqueue = new LinkedBlockingQueue<Notification>(); // 待接收投票队列
        // 经过QuorumCnxManager往sendqueue添加投票，从recvqueue获取投票
        this.messenger = new Messenger(manager);
    }
    ...
    public Vote lookForLeader() throws InterruptedException {
        ...
        try {
 HashMap<Long, Vote> recvset = new HashMap<Long, Vote>(); // 投票箱 // 如当前ZK服务器新加入到集群时，选举已经结束且当前选举轮数落后
            HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();
            ...
            synchronized(this){
                logicalclock++; // 选举轮数 + 1
                // 初始推荐Leader为本身，即初始投票投给本身
                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
            }
            ...
 sendNotifications(); // 广播本身的投票 // 循环直至选举出Leader
            while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){
                // 获取一个其它ZK服务器的投票（超时时间为200毫秒）
                Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS); if(n == null){
                    // 从新广播本身的投票，增长超时时间
                    ...
                }
                else if(self.getVotingView().containsKey(n.sid)) {
                    switch (n.state) {
                    case LOOKING: // 选举进行中
                        if (n.electionEpoch > logicalclock) { // 当前选举轮数落后
                            logicalclock = n.electionEpoch; // 更新选举轮数
                            recvset.clear(); // 清空票箱
                            // 对方投票 > 当前投票（对方推荐的LeaderID > 本身推荐的LeaderID）
                            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) { // 当前选举轮数领先
                            // 纪录日志，不作其它处理
                            ...
                            break;
                        } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {
                            updateProposal(n.leader, n.zxid, n.peerEpoch); // 跟随对方投票
 sendNotifications(); // 从新广播本身的投票
                        }
                        ...
                        // 更新票箱
                        recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch)); // 对方投票过半
                        if (termPredicate(recvset, new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch))) { // 等待200ms接收投票
                            while((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
                                // 对方投票 > 当前投票，则选举继续
                                if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)){
                                    recvqueue.put(n);
                                    break;
                                }
                            }
                            if (n == null) { // 200ms内未接收到新的投票
                                // 服务器状态由LOOKING转为LEADING/FOLLOWING/OBSERVING
                                self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING: learningState()); // 确认和返回最终投票
                                Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch);                                 leaveInstance(endVote);
                                return endVote;                             }
                        }
                        break;
                    case OBSERVING: break;
                    case FOLLOWING:  case LEADING: // 如当前ZK服务器新加入到集群时，选举已经结束
                        if(n.electionEpoch == logicalclock) { // ??为什么选举轮数落后则使用outofelection
                            // 更新recvset票箱，并查找当前集群Leader
                            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;
                            }
                        }
                        // 更新outofelection票箱，并查找当前集群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 = n.electionEpoch;
                                self.setPeerState((n.leader == self.getId()) ? erverState.LEADING: learningState());
                            }
                            Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
                            leaveInstance(endVote);
                            return endVote;
                        }
                        break;
                    ...
                    }
                }
                ...
            }
            return null;
        }
        ...
    }
}