Zookeeper启动和集群选举

ゞ浴缸里的玫瑰 2021-12-24 03:31 311阅读 0赞

## 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;
            }
            ...
        }
    }

转载于:https://www.cnblogs.com/bjorney/p/10443178.html