第六章 ZK集群源码解读笔记

一时失言乱红尘 2024-04-17 06:03 68阅读 0赞

## **一、集群模式** ##

### **1、数据同步总流程** ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70][]

1）进入QuorumPeerMain .main()方法

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

2）QuorumPeerMain.runFromConfig(config)方法

if (args.length == 1 && config.servers.size() > 0) {
        runFromConfig(config); // 集群模式
    } else { // there is only server in the quorum -- run as standalone
        ZooKeeperServerMain.main(args);
    }
    
    public void runFromConfig(QuorumPeerConfig config) throws IOException {
      try { 
    
    //创建NIOServerCnxnFactory：serverCnxnFactoryName = NIOServerCnxnFactory.class.getName(); 
          ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
    
    // NIOServerCnxnFactory.configure()
          cnxnFactory.configure(config.getClientPortAddress(),config.getMaxClientCnxns());
          quorumPeer = getQuorumPeer();
          quorumPeer.setQuorumPeers(config.getServers());
          quorumPeer.setTxnFactory(new FileTxnSnapLog(new File(config.getDataLogDir()),
    
    new File(config.getDataDir())));
          quorumPeer.setElectionType(config.getElectionAlg());
          quorumPeer.setMyid(config.getServerId());
          quorumPeer.setTickTime(config.getTickTime());
          quorumPeer.setInitLimit(config.getInitLimit());
          quorumPeer.setSyncLimit(config.getSyncLimit());
          ......
          quorumPeer.setQuorumCnxnThreadsSize(config.quorumCnxnThreadsSize);

**3）初始化quorumPeer并启动**

quorumPeer.initialize();
          quorumPeer.start();
          quorumPeer.join();
      } catch (InterruptedException e) { // warn, but generally this is ok
          LOG.warn("Quorum Peer interrupted", e);
      }
    }
    
    @Override
    public void configure(InetSocketAddress addr, int maxcc) throws IOException {
        configureSaslLogin();
        thread = new ZooKeeperThread(this, "NIOServerCxn.Factory:" + addr);
        thread.setDaemon(true);
        maxClientCnxns = maxcc;
        this.ss = ServerSocketChannel.open();
        ss.socket().setReuseAddress(true);
        LOG.info("binding to port " + addr);
        ss.socket().bind(addr);
        ss.configureBlocking(false);
        ss.register(selector, SelectionKey.OP_ACCEPT);
    }
    
    @Override
    public synchronized void start() {
        loadDataBase();  // 将快照信息加载到内存
        cnxnFactory.start();  // 构建读写request
        startLeaderElection(); // 集群选举
        super.start(); //实际调用QuorumPeer.run()方法
    }

**4）loadDataBase()把快照加载到内存**

private void loadDataBase() {
           File updating = new File(getTxnFactory().getSnapDir(),UPDATING_EPOCH_FILENAME);
           try {
              zkDb.loadDataBase();
              long lastProcessedZxid = zkDb.getDataTree().lastProcessedZxid;
              long epochOfZxid = ZxidUtils.getEpochFromZxid(lastProcessedZxid);
               try {
                   currentEpoch = readLongFromFile(CURRENT_EPOCH_FILENAME);
                   if (epochOfZxid > currentEpoch && updating.exists()) {
                       setCurrentEpoch(epochOfZxid);
                   }
               } 
               try {
                   acceptedEpoch = readLongFromFile(ACCEPTED_EPOCH_FILENAME);
               }
           }
    }

5）NIOServerCnxnFactory.start();  前面介绍过该方法，通过run()方法构建读写的request，

6）进行集群选举startLeaderElection();

7）执行super.start(); 实际调用QuorumPeer.run()方法

8）判断服务器的状态进行初始化QuorumPeer.run()

case LOOKING:
        LOG.info("LOOKING");
    case OBSERVING:
            LOG.info("OBSERVING");
            setObserver(makeObserver(logFactory));
            observer.observeLeader();
    case FOLLOWING:
            setFollower(makeFollower(logFactory));
            follower.followLeader();
    case LEADING:
            setLeader(makeLeader(logFactory));
            leader.lead();
            setLeader(null);

**1.1、OBSERVING**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 1][]

2）创建ObserverZooKeeperServer

protected Observer makeObserver(FileTxnSnapLog logFactory) throws IOException {
        return new Observer(this, new ObserverZooKeeperServer(logFactory,
                this, new ZooKeeperServer.BasicDataTreeBuilder(), this.zkDb));
    }

3）以observer.observeLeader();方式与leader交互

4）通过findLeader()方法找到leader服务器

void observeLeader() throws InterruptedException {
        zk.registerJMX(new ObserverBean(this, zk), self.jmxLocalPeerBean);
        try {
            QuorumServer leaderServer = findLeader();
            try {
                connectToLeader(leaderServer.addr, leaderServer.hostname);
                long newLeaderZxid = registerWithLeader(Leader.OBSERVERINFO);
                syncWithLeader(newLeaderZxid);
                QuorumPacket qp = new QuorumPacket();
                while (this.isRunning()) {
                    readPacket(qp);
                    processPacket(qp);                   
                }
            } 
        } finally {
            zk.unregisterJMX(this);
        }
    }

**5）从leader同步并调用zk.startup()方法**

protected void syncWithLeader(long newLeaderZxid) throws IOException, InterruptedException{
        QuorumPacket ack = new QuorumPacket(Leader.ACK, 0, null, null);
        QuorumPacket qp = new QuorumPacket();
        ......
    
    zk.startup();

**6）调用ObserverZooKeeperServer.setupRequestProcessors()方法**

protected void setupRequestProcessors() {      
        RequestProcessor finalProcessor = new FinalRequestProcessor(this);
        commitProcessor = new CommitProcessor(finalProcessor, Long.toString(getServerId()), true,getZooKeeperServerListener());
        commitProcessor.start();
        firstProcessor = new ObserverRequestProcessor(this, commitProcessor);
        ((ObserverRequestProcessor) firstProcessor).start();
        if (syncRequestProcessorEnabled) {
            syncProcessor = new SyncRequestProcessor(this, null);
            syncProcessor.start();
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 2][]

**7）通过ObserverRequestProcessor.processRequest将请求加入到queuedRequests队列中**

**//根据构造方法传入的实际processor，这里传入的就是ObserverRequestProcessor**  
**nextProcessor****.processRequest(request);**

public void processRequest(Request request) {
        if (!finished) {
            queuedRequests.add(request);
        }
    }

8）执行线程逻辑，从queuedRequests.take()中取出request，通过 zks.getObserver().request(request);方法，如果是事务请求就把request包装成Leader.RQUEST类型的QuorumPacket发送给leader

@Override
    public void run() {
        try {
            while (!finished) {
                Request request = queuedRequests.take();
                if (request == Request.requestOfDeath) {
                    break;
                }
                nextProcessor.processRequest(request);
                switch (request.type) {
                case OpCode.sync:
                    zks.pendingSyncs.add(request);
                    zks.getObserver().request(request);
                    break;
                case OpCode.create:
                case OpCode.delete:
                case OpCode.setData:
                case OpCode.setACL:
                case OpCode.createSession:
                case OpCode.closeSession:
                case OpCode.multi:
                    zks.getObserver().request(request);
                    break;
                }
            }
        }
    }
    
    QuorumPacket qp = new QuorumPacket(Leader.REQUEST, -1, baos.toByteArray(), request.authInfo);
    writePacket(qp, true);
    
    void writePacket(QuorumPacket pp, boolean flush) throws IOException {
        synchronized (leaderOs) {
            if (pp != null) {
                leaderOs.writeRecord(pp, "packet");
            }
            if (flush) {
                bufferedOutput.flush();
            }
        }
    }

9）nextProcessor.processRequest(request)方法实际调用CommitProcessor.processRequest(request)，如果是非事务请求，直接交给后面的FinalRequestProcessor处理，如果是事务请求，等待leader发送commit请求被唤醒，再把运行提交的事务请求添加都FinalRequestProcessor处理。

下面为CommitProcessor的run方法代码：

@Override
    public void run() {
        try {
            Request nextPending = null;            
            while (!finished) {
                int len = toProcess.size();
                for (int i = 0; i < len; i++) {
                    nextProcessor.processRequest(toProcess.get(i));
                }
                toProcess.clear();
                synchronized (this) {
                    if ((queuedRequests.size() == 0 || nextPending != null) && committedRequests.size() == 0) {
                        wait();
                        continue;
                    }
                    if ((queuedRequests.size() == 0 || nextPending != null)&& committedRequests.size() > 0) {
                        Request r = committedRequests.remove();
                       if(nextPending!= null && nextPending.sessionId==r.sessionId && nextPending.cxid == r.cxid){
                            nextPending.hdr = r.hdr;
                            nextPending.txn = r.txn;
                            nextPending.zxid = r.zxid;
                            toProcess.add(nextPending);
                            nextPending = null;
                        } else {
                            toProcess.add(r);
                        }
                    }
                }
                if (nextPending != null) {
                    continue;
                }
                synchronized (this) {// Process the next requests in the queuedRequests
                    while (nextPending == null && queuedRequests.size() > 0) {
                        Request request = queuedRequests.remove();
                        switch (request.type) {
                        case OpCode.create:
                        case OpCode.delete:
                        case OpCode.setData:
                        case OpCode.multi:
                        case OpCode.setACL:
                        case OpCode.createSession:
                        case OpCode.closeSession:
                            nextPending = request;
                            break;
                        case OpCode.sync:
                            if (matchSyncs) {
                                nextPending = request;
                            } else {
                                toProcess.add(request);
                            }
                            break;
                        default:
                            toProcess.add(request);
                        }
                    }
                }
            }
        } 
        LOG.info("CommitProcessor exited loop!");
    }

**10）最后FinalRequestProcessor把处理结果返回，具体处理方法参考代码如下：**

public void processRequest(Request request) {
        ......
        ProcessTxnResult rc = null;
        if (request.cnxn == null) { return; }
        ServerCnxn cnxn = request.cnxn;
        String lastOp = "NA";
        zks.decInProcess();
        Code err = Code.OK;
        Record rsp = null;
        boolean closeSession = false;
        try {
            KeeperException ke = request.getException();
            if (ke != null && request.type != OpCode.multi) {
                throw ke;
            }
            switch (request.type) {
            ......
            case OpCode.multi: {
                lastOp = "MULT";
                rsp = new MultiResponse() ;
                for (ProcessTxnResult subTxnResult : rc.multiResult) {
                    OpResult subResult ;
                    switch (subTxnResult.type) {
                        case OpCode.check:
                            subResult = new CheckResult();
                            break;
                        case OpCode.create:
                            subResult = new CreateResult(subTxnResult.path);
                            break;
                        case OpCode.delete:
                            subResult = new DeleteResult();
                            break;
                        case OpCode.setData:
                            subResult = new SetDataResult(subTxnResult.stat);
                            break;
                        case OpCode.error:
                            subResult = new ErrorResult(subTxnResult.err) ;
                            break;
                        default:
                            throw new IOException("Invalid type of op");
                    }
                    ((MultiResponse)rsp).add(subResult);
                }
                break;
            }
            case OpCode.create: {
                lastOp = "CREA";
                rsp = new CreateResponse(rc.path);
                err = Code.get(rc.err);
                break;
            }
            case OpCode.delete: {
                lastOp = "DELE";
                err = Code.get(rc.err);
                break;
            }
            case OpCode.setData: {
                lastOp = "SETD";
                rsp = new SetDataResponse(rc.stat);
                err = Code.get(rc.err);
                break;
            }
            case OpCode.getData: {
                lastOp = "GETD";
                GetDataRequest getDataRequest = new GetDataRequest();
                ByteBufferInputStream.byteBuffer2Record(request.request,getDataRequest);
                DataNode n = zks.getZKDatabase().getNode(getDataRequest.getPath());
          PrepRequestProcessor.checkACL(zks, zks.getZKDatabase().aclForNode(n),ZooDefs.Perms.READ,request.authInfo);
                Stat stat = new Stat();
        byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,getDataRequest.getWatch() ? cnxn : null);
                rsp = new GetDataResponse(b, stat);
                break;
            }
            ......
            }
        } 
        long lastZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
        ReplyHeader hdr =new ReplyHeader(request.cxid, lastZxid, err.intValue());
        zks.serverStats().updateLatency(request.createTime);
        cnxn.updateStatsForResponse(request.cxid, lastZxid, lastOp, request.createTime, Time.currentElapsedTime());
        try {
            cnxn.sendResponse(hdr, rsp, "response");
            if (closeSession) {
                cnxn.sendCloseSession();
            }
        } 
    }

**1.2、FOLLOWING**

具体代码流程与上面的观察者模式类似，这里不再详述

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 4][]

![20190825234235666.png][]

**1.3、LEADING处理流程**

**1）流程入口与前面的都一样**

**2）makeLeader方法创建ZooKeeperServer**

protected Leader makeLeader(FileTxnSnapLog logFactory) throws IOException {
        return new Leader(this, new LeaderZooKeeperServer(logFactory,this,new ZooKeeperServer.BasicDataTreeBuilder(), this.zkDb));
    }
    
    static public class BasicDataTreeBuilder implements DataTreeBuilder {
        public DataTree build() {
            return new DataTree();
        }
    }

**3）leader.lead()方法完成leader的职责**

void lead() throws IOException, InterruptedException {
        self.end_fle = Time.currentElapsedTime();
        long electionTimeTaken = self.end_fle - self.start_fle;
        self.setElectionTimeTaken(electionTimeTaken);
        self.start_fle = 0;
        self.end_fle = 0;
        zk.registerJMX(new LeaderBean(this, zk), self.jmxLocalPeerBean);
        try {
            self.tick.set(0);
            zk.loadData();
            leaderStateSummary = new StateSummary(self.getCurrentEpoch(), zk.getLastProcessedZxid());

** 4） 创建并运行LearnerCnxAcceptor线程（对每一个learner开启一个线程），根据learner的情况同步Leader信息，等待learner发送的Packet。**

// Start thread that waits for connection requests from new followers
            cnxAcceptor = new LearnerCnxAcceptor();
            cnxAcceptor.start();
            readyToStart = true;
            long epoch = getEpochToPropose(self.getId(), self.getAcceptedEpoch());
            zk.setZxid(ZxidUtils.makeZxid(epoch, 0));
            synchronized(this){
                lastProposed = zk.getZxid();
            }
            newLeaderProposal.packet = new QuorumPacket(NEWLEADER, zk.getZxid(), null, null);
            if ((newLeaderProposal.packet.getZxid() & 0xffffffffL) != 0) {
                LOG.info("NEWLEADER proposal has Zxid of "+      Long.toHexString(newLeaderProposal.packet.getZxid()));
            }
            waitForEpochAck(self.getId(), leaderStateSummary);
            self.setCurrentEpoch(epoch);
            // We have to get at least a majority of servers in sync with us. We do this by waiting for
    
    // the NEWLEADER packet to get acknowledged
            try {
                waitForNewLeaderAck(self.getId(), zk.getZxid(), LearnerType.PARTICIPANT);
            } catch (InterruptedException e) {
                ......
            }
    
    // 5）运行LeaderZooKeeperServer服务器
            startZkServer();
            String initialZxid = System.getProperty("zookeeper.testingonly.initialZxid");
            if (initialZxid != null) {
                long zxid = Long.parseLong(initialZxid);
                zk.setZxid((zk.getZxid() & 0xffffffff00000000L) | zxid);
            }
            if (!System.getProperty("zookeeper.leaderServes", "yes").equals("no")) {
                self.cnxnFactory.setZooKeeperServer(zk);
            }
            boolean tickSkip = true;
            while (true) {
                Thread.sleep(self.tickTime / 2);
                if (!tickSkip) {  self.tick.incrementAndGet(); }
                HashSet<Long> syncedSet = new HashSet<Long>();
                // lock on the followers when we use it.
                syncedSet.add(self.getId());
                for (LearnerHandler f : getLearners()) {
                    // Synced set is used to check we have a supporting quorum, so only
                    // PARTICIPANT, not OBSERVER, learners should be used
                    if (f.synced() && f.getLearnerType() == LearnerType.PARTICIPANT) {
                        syncedSet.add(f.getSid());
                    }
                    f.ping();
                }
                if (!this.isRunning()) {// check leader running status
                    shutdown("Unexpected internal error");
                    return;
                }
              if (!tickSkip && !self.getQuorumVerifier().containsQuorum(syncedSet)) {
              shutdown("Not sufficient followers synced, only synced with sids: [ " + getSidSetString(syncedSet) + " ]");
                    return; // make sure the order is the same!the leader goes to looking
              } 
              tickSkip = !tickSkip;
            }
        } finally {
            zk.unregisterJMX(this);
        }
    }

**5）运行LeaderZooKeeperServer服务器**

// Start up Leader ZooKeeper server and initialize zxid to the new epoch
    private synchronized void startZkServer() {
        lastCommitted = zk.getZxid();
        zk.startup();
        // Update the election vote here to ensure that all members of theensemble report the same vote to new
    
    // servers that start up and send leader election notifications to the ensemble.
        // @see https://issues.apache.org/jira/browse/ZOOKEEPER-1732
        self.updateElectionVote(getEpoch()); // 更新选举投票
        zk.getZKDatabase().setlastProcessedZxid(zk.getZxid()); // 更新zxid
    }

6）调用ZooKeeperServer.setupRequestProcessors()方法处理请求

public synchronized void startup() {
        if (sessionTracker == null) {
            createSessionTracker();
        }
        startSessionTracker();
        setupRequestProcessors();
        registerJMX();
        setState(State.RUNNING);
        notifyAll();
    }

7）SyncRequestProcessor、PrepRequestProcessor、FinalRequestProcessor处理请求并返回结果

protected void setupRequestProcessors() {
        RequestProcessor finalProcessor = new FinalRequestProcessor(this);
        RequestProcessor syncProcessor = new SyncRequestProcessor(this,finalProcessor);
        ((SyncRequestProcessor)syncProcessor).start();
        firstProcessor = new PrepRequestProcessor(this, syncProcessor);
        ((PrepRequestProcessor)firstProcessor).start();
    }

具体各种processor处理流程如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 5][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 6][]

**1）调用PrepRequestProcessor.processRequest()方法将请求加入到submittedRequests队列中**

public void processRequest(Request request) {// request.addRQRec(">prep="+zks.outstandingChanges.size());
        submittedRequests.add(request);
    }

**2）通过ProposalRequestProcessor.processRequest()方法处理request请求。**

public void run() {
        try {
            while (true) {
                Request request = submittedRequests.take();
                long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
                if (request.type == OpCode.ping) {
                    traceMask = ZooTrace.CLIENT_PING_TRACE_MASK;
                }
                if (Request.requestOfDeath == request) {  break; }
                pRequest(request);
            }
        } 
    }
    
    protected void pRequest(Request request) throws RequestProcessorException {
        request.hdr = null;
        request.txn = null;
        try {
            switch (request.type) {
    ......
    
    request.zxid = zks.getZxid();
    nextProcessor.processRequest(request);

**ProposalRequestProcessor.processRequest()处理request请求**

> **public void** processRequest(Request request) **throws** RequestProcessorException \{   
> *    ***if**(request **instanceof** LearnerSyncRequest)\{ 
> 
> **//1、 如果接受的是learner发送的同步请求**  
> **        zks.getLeader().processSync((LearnerSyncRequest)request);**  
>     \} **else** \{ ** //2、 如果是其他请求**  
>             **nextProcessor**.processRequest(request);  
>         **if** (request.**hdr** != **null**) \{ *// We need to sync and get consensus on any transactions*  
> *            ***try** \{   
>                 **zks.getLeader().propose(request);**  
>             \} **catch** (XidRolloverException e) \{   
>                 **throw new** RequestProcessorException(e.getMessage(), e);  
>             \}  
>             **syncProcessor**.processRequest(request);  
>         \}  
>     \}  
> \}

如果是同步：Leader().processSync()方法将请求封装成Leader.SYNC的Packet并且加入到队列QuorumPacket中。

synchronized public void processSync(LearnerSyncRequest r){
        if(outstandingProposals.isEmpty()){
            sendSync(r);
        } else {
            List<LearnerSyncRequest> l = pendingSyncs.get(lastProposed);
            if (l == null) {
                l = new ArrayList<LearnerSyncRequest>();
            }
            l.add(r);
            pendingSyncs.put(lastProposed, l);
        }
    }
    
    public void sendSync(LearnerSyncRequest r){
        QuorumPacket qp = new QuorumPacket(Leader.SYNC, 0, null, null);
        r.fh.queuePacket(qp);
    }

4）如果是其他请求：nextProcessor.processRequest(request)实际执行的是CommitProcessor.

processRequest(request)处理逻辑，如果是非事务 请求直接交给后面的ToBeAppliedRequestProcessor去处理，如果是事务请求则等待Leader投票发送commit请求被唤醒，在吧运行提交的事务添加到ToBeAppliedRequestProcessor中去处理。

**CommitProcessor****.**run();

public void run() {
        try {
            Request nextPending = null;            
            while (!finished) {
                int len = toProcess.size();
                for (int i = 0; i < len; i++) {
                    nextProcessor.processRequest(toProcess.get(i));
                }
                toProcess.clear();
                ......

**ToBeAppliedRequestProcessor.processRequest(request)将投票信息删除，最后调用FinalRequestProcessor**

返回response结果。

public void processRequest(Request request) throws RequestProcessorException {
        next.processRequest(request); // 调用FinalRequestProcessor返回结果
        Proposal p = toBeApplied.peek();
        if (p != null && p.request != null && p.request.zxid == request.zxid) {
            toBeApplied.remove();
        }
    }

5）继续第2步ProposalRequestProcessor.processRequest()处理剩下的逻辑： zks.getLeader().propose(request)发起投票处理

public Proposal propose(Request request) throws XidRolloverException {
        if ((request.zxid & 0xffffffffL) == 0xffffffffL) {
            String msg = "zxid lower 32 bits have rolled over, forcing re-election, and therefore new epoch start";
            shutdown(msg);
            throw new XidRolloverException(msg);
        }
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
        try {
            request.hdr.serialize(boa, "hdr");
            if (request.txn != null) {
                request.txn.serialize(boa, "txn");
            }
            baos.close();
        } catch (IOException e) {
            LOG.warn("This really should be impossible", e);
        }
        QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, baos.toByteArray(), null);
        Proposal p = new Proposal();
        p.packet = pp;
        p.request = request;
        synchronized (this) {
            lastProposed = p.packet.getZxid();
            outstandingProposals.put(lastProposed, p);
            sendPacket(pp);
        }
        return p;
    }
    
    void sendPacket(QuorumPacket qp) {
        synchronized (forwardingFollowers) {
            for (LearnerHandler f : forwardingFollowers) {                
                f.queuePacket(qp);
            }
        }
    }

### **2、领导选举** ###

**2.1、领导选举初始化**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 7][]

前五步之前的逻辑都介绍过了，下面从第六步开始梳理：

> @Override  
> **public synchronized void** start() \{   
>     loadDataBase();  
>     **cnxnFactory**.start();          
>     **startLeaderElection();**** // 开始Leader选举**  
>     **super**.start();  
> \}

7）创建投票，投票给自己然后QuorumPeer.createElectionAlgorith()方法创建选举策略，选举策略根据electionType确定，其取值默认为QuorumPeerConfig.electionAlg = 3;

synchronized public void startLeaderElection() {
       try {   currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch()); }
       for (QuorumServer p : getView().values()) {
            if (p.id == myid) {
                myQuorumAddr = p.addr;
                break;
            }
        }
        if (electionType == 0) {
            try {
                udpSocket = new DatagramSocket(myQuorumAddr.getPort());
                responder = new ResponderThread();
                responder.start();
            } catch (SocketException e) {  throw new RuntimeException(e); }
        }
        this.electionAlg = createElectionAlgorithm(electionType);
    }
    
    protected Election createElectionAlgorithm(int electionAlgorithm){
        Election le=null;
        switch (electionAlgorithm) {
        case 0:
            le = new LeaderElection(this);
            break;
        case 1:
            le = new AuthFastLeaderElection(this);
            break;
        case 2:
            le = new AuthFastLeaderElection(this, true);
            break;
        case 3:
            qcm = createCnxnManager();
            QuorumCnxManager.Listener listener = qcm.listener;
            if(listener != null){
                listener.start();
                le = new FastLeaderElection(this, qcm);
            } 
            break;
        default:
            assert false;
        }
        return le;
    }

**8）创建QuorumCnxManager初始化：**

*  senderWorkerMap：每个服务sid对应的sendWorker
 *  recvQueue：本台服务器接收的消息
 *  queueSendMap：需要发送给每个服务器sid的消息队列
 *  lastMessageSent：发送给每台服务器的最近消息

public QuorumCnxManager(final long mySid, Map<Long,QuorumPeer.QuorumServer> view,QuorumAuthServer authServer,
           QuorumAuthLearner authLearner,int socketTimeout, boolean listenOnAllIPs,int quorumCnxnThreadsSize,
           boolean quorumSaslAuthEnabled,ConcurrentHashMap<Long, SendWorker> senderWorkerMap) {
        this.senderWorkerMap = senderWorkerMap;
        this.recvQueue = new ArrayBlockingQueue<Message>(RECV_CAPACITY);
        this.queueSendMap = new ConcurrentHashMap<Long, ArrayBlockingQueue<ByteBuffer>>();
        this.lastMessageSent = new ConcurrentHashMap<Long, ByteBuffer>();
        String cnxToValue = System.getProperty("zookeeper.cnxTimeout");
        if(cnxToValue != null){
            this.cnxTO = Integer.parseInt(cnxToValue);
        }
        this.mySid = mySid;
        this.socketTimeout = socketTimeout;
        this.view = view;
        this.listenOnAllIPs = listenOnAllIPs;
        initializeAuth(mySid, authServer, authLearner, quorumCnxnThreadsSize, quorumSaslAuthEnabled);
        // Starts listener thread that waits for connection requests 
        listener = new Listener();
    }

**9）继续QuorumPeer.createElectionAlgorith()方法，初始化FastLeaderElection(this, qcm)，调用**

public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager){
        this.stop = false;
        this.manager = manager;
        starter(self, manager); // 初始化发送和接收对象
    }
    
    private void starter(QuorumPeer self, QuorumCnxManager manager) {
        this.self = self;
        proposedLeader = -1;
        proposedZxid = -1;
        sendqueue = new LinkedBlockingQueue<ToSend>(); // 需要发送给每个服务器的消息队列
        recvqueue = new LinkedBlockingQueue<Notification>(); // 接收到的Notification对象
        this.messenger = new Messenger(manager);
    }

**10）创建Messenger对象，在里面创建并启动WorkerSender和WorkerReceiver线程：**

FastLeaderElection.Messenger()

Messenger(QuorumCnxManager manager) {
        this.ws = new WorkerSender(manager);
        Thread t = new Thread(this.ws,"WorkerSender[myid=" + self.getId() + "]");
        t.setDaemon(true);
        t.start();
        this.wr = new WorkerReceiver(manager);
        t = new Thread(this.wr, "WorkerReceiver[myid=" + self.getId() + "]");
        t.setDaemon(true);
        t.start();
    }

WorkerSender的run方法负责从sendqueue中取出数据，并添加到queueSendMap.putIfAbsent(sid, bq);队列中，然后通过QuorumCnxManager.toSend()方法使用socket方式发送给其他服务器。

class WorkerSender extends ZooKeeperThread {
        volatile boolean stop;
        QuorumCnxManager manager;
        WorkerSender(QuorumCnxManager manager){
            super("WorkerSender");
            this.stop = false;
            this.manager = manager;
        }
        public void run() {
            while (!stop) {
                try {
                    ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
                    if(m == null) continue;
                    process(m);
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
        void process(ToSend m) {
            ByteBuffer requestBuffer = buildMsg(m.state.ordinal(), m.leader,m.zxid, m.electionEpoch, m.peerEpoch);
            manager.toSend(m.sid, requestBuffer);
        }
    }

WorkerReceiver的run方法负责从QuorumCnxManager.recvQueue中获取数据，然后进行相应的处理。

class WorkerReceiver extends ZooKeeperThread {
        volatile boolean stop;
        QuorumCnxManager manager;
        WorkerReceiver(QuorumCnxManager manager) {
            super("WorkerReceiver");
            this.stop = false;
            this.manager = manager;
        }
        public void run() {
            Message response;
            while (!stop) { // Sleeps on receive
                try{
                    response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
                    if(response == null) continue;
                    /*
                     * If it is from an observer, respond right away.Note that the following predicate assumes that
                     * if a server is not a follower, then it must be an observer. If we ever have any other type of
                     * learner in the future, we'll have to change the way we check for observers.
                     */
                    if(!self.getVotingView().containsKey(response.sid)){
                        Vote current = self.getCurrentVote();
                        ToSend notmsg = new ToSend(ToSend.mType.notification, current.getId(), current.getZxid(),
                                logicalclock.get(),self.getPeerState(),response.sid,current.getPeerEpoch());
                        sendqueue.offer(notmsg);
                    } else {// Receive new message
                        if (response.buffer.capacity() < 28) {
                            continue;
                        }
                        boolean backCompatibility = (response.buffer.capacity() == 28);
                        response.buffer.clear();
                        Notification n = new Notification(); // Instantiate Notification and set its attributes
                        // State of peer that sent this message
                        QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
                        switch (response.buffer.getInt()) {
                        case 0:
                            ackstate = QuorumPeer.ServerState.LOOKING;
                            break;
                        case 1:
                            ackstate = QuorumPeer.ServerState.FOLLOWING;
                            break;
                        case 2:
                            ackstate = QuorumPeer.ServerState.LEADING;
                            break;
                        case 3:
                            ackstate = QuorumPeer.ServerState.OBSERVING;
                            break;
                        default:
                            continue;
                        }
                        n.leader = response.buffer.getLong();
                        n.zxid = response.buffer.getLong();
                        n.electionEpoch = response.buffer.getLong();
                        n.state = ackstate;
                        n.sid = response.sid;
                        if(!backCompatibility){
                            n.peerEpoch = response.buffer.getLong();
                        } else {
                            n.peerEpoch = ZxidUtils.getEpochFromZxid(n.zxid);
                        }
                        n.version = (response.buffer.remaining() >= 4) ?  response.buffer.getInt() : 0x0;
                        // if this server is looking, then send proposed leader
                        if(self.getPeerState() == QuorumPeer.ServerState.LOOKING){
                            recvqueue.offer(n);
                        if((ackstate == QuorumPeer.ServerState.LOOKING) && (n.electionEpoch < logicalclock.get())){
                                Vote v = getVote();
                                ToSend notmsg = new       ToSend(ToSend.mType.notification,v.getId(),v.getZxid(),
                                        logicalclock.get(),self.getPeerState(),response.sid, v.getPeerEpoch());
                                sendqueue.offer(notmsg);
                            }
                        } else {
                            Vote current = self.getCurrentVote();
                            if(ackstate == QuorumPeer.ServerState.LOOKING){
                                ToSend notmsg;
                                if(n.version > 0x0) {
              notmsg = new ToSend(ToSend.mType.notification, current.getId(),current.getZxid(),
    
    current.getElectionEpoch(),self.getPeerState(),response.sid,current.getPeerEpoch());
                                } else {
                                    Vote bcVote = self.getBCVote();
                                    notmsg = new     ToSend(ToSend.mType.notification,bcVote.getId(),bcVote.getZxid(),
           bcVote.getElectionEpoch(),self.getPeerState(),response.sid,bcVote.getPeerEpoch());
                                }
                                sendqueue.offer(notmsg);
                                  ......
    }

**2.2、线程逻辑**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 8][]

**1）发送请求用到的线程**

public void toSend(Long sid, ByteBuffer b) {
        if (this.mySid == sid) {
             b.position(0);
             addToRecvQueue(new Message(b.duplicate(), sid));
        } else {
             ArrayBlockingQueue<ByteBuffer> bq = new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY);
             ArrayBlockingQueue<ByteBuffer> bqExisting = queueSendMap.putIfAbsent(sid, bq);
             if (bqExisting != null) {  addToSendQueue(bqExisting, b);
             } else {  addToSendQueue(bq, b);  }
             connectOne(sid); 
        }
    }
    
    // Try to establish a connection to server with id sid.
    synchronized public void connectOne(long sid){
        if (!connectedToPeer(sid)){
            InetSocketAddress electionAddr;
            if (view.containsKey(sid)) {
                electionAddr = view.get(sid).electionAddr;
            } 
            try {
                LOG.debug("Opening channel to server " + sid);
                Socket sock = new Socket();
                setSockOpts(sock);
                sock.connect(view.get(sid).electionAddr, cnxTO);
                LOG.debug("Connected to server " + sid);
                if (quorumSaslAuthEnabled) {
                    initiateConnectionAsync(sock, sid);
                } else {
                    initiateConnection(sock, sid);
                }
            }
    
            ......
        } else {
            LOG.debug("There is a connection already for server " + sid);
        }
    }

**2）接收请求用到的线程**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 9][]

**public** Message pollRecvQueue(**long** timeout, TimeUnit unit)**throws** InterruptedException \{   
   **return** **recvQueue**.poll(timeout, unit);  
\}

**3）发送/接收总览**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 10][]

**2.3、领导选举算法**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 11][]

*  保存信息到recvset
 *  termPredicate方法自己投票的那台 服务器不能当领导（过半机器）
 *  如果选出了Leader，那么还要判断在这个时间段内是否又来了新的投票，很可能需要重新选举，如果新的投票比选举出开的服务器更优秀，则重新选举，知道没有获取新的结果。
 *  最后比较最终的结果是不是自己，如果是则更新为leader不是则更新为follower。

**1）调用QuorumServer.run()方法进入选举流程**

> @Override  
> **public synchronized void** start() \{   
>     loadDataBase();  
>     **cnxnFactory**.start();          
>     startLeaderElection();  
>     **super**.start(); // 启动线程，执行run方法  
> \}

2）执行run方法进入LOOKING分支

switch (getPeerState()) {
    case LOOKING:
        if (Boolean.getBoolean("readonlymode.enabled")) {
            // Create read-only server but don't start it immediately
            final ReadOnlyZooKeeperServer roZk = new ReadOnlyZooKeeperServer( logFactory,this,
                  new ZooKeeperServer.BasicDataTreeBuilder(),this.zkDb);
            Thread roZkMgr = new Thread() {
                public void run() {
                    try { // lower-bound grace period to 2 secs
                        sleep(Math.max(2000, tickTime));
                        if (ServerState.LOOKING.equals(getPeerState())) {
                            roZk.startup();
                        }
                    } catch (InterruptedException e) {
                        LOG.info("Interrupted while attempting to start ReadOnlyZooKeeperServer, not started");
                    } 
                }
            };
            try {
                roZkMgr.start();
                setBCVote(null);
                setCurrentVote(makeLEStrategy().lookForLeader());  // 选举leader
            } catch (Exception e) {
                setPeerState(ServerState.LOOKING);
            } finally {
                roZkMgr.interrupt();
                roZk.shutdown();
            }
        } else {
            try {
                setBCVote(null);
                setCurrentVote(makeLEStrategy().lookForLeader());
            } catch (Exception e) {
                LOG.warn("Unexpected exception", e);
                setPeerState(ServerState.LOOKING);
            }
        }
        break;
    case OBSERVING:

3）调用QuorumPeer.setCurrentVote(makeLEStrategy().lookForLeader());

4）实际调用的是FastLeaderElection.lookForLeader()方法

5）启动时先投票给自己（sendNotifications()加入到sendqueue.offer(notmsg)中）

> **synchronized**(**this**)\{   
>     **logicalclock**.incrementAndGet();  
>     updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());  
> \}  
> sendNotifications();
> 
>     sendqueue.offer(notmsg);

6）从recvqueue队里中获取数据（WorkerReceiver线程收到message，然后把message封装成Notification加入到recvqueue队列中）。

> **while** ((**self**.getPeerState() == ServerState.***LOOKING***) && (!**stop**))\{   
>     */**/* *Remove next notification from queue, times out after 2 times** **the termination time*  
> *    *Notification n = **recvqueue**.poll(notTimeout,TimeUnit.***MILLISECONDS***);

7）如果获取数据可以投票（不是observer）则进行状态判定switch (n.state)

8）如果是observer则什么都不做；

如果是LEADING、FOLLOWING，并且选票是自己，则把自己的状态修改为LEADING；

如果状态还是looking，说明现在还没有选出领导。

> **switch** (n.**state**) \{ 
> 
> **case** ***LOOKING***:
> 
>     ......
> 
> **case** ***OBSERVING***:  
>         ***LOG***.debug(**"Notification from observer: "** \+ n.**sid**);  
>         **break**;  
>     **case** ***FOLLOWING***:  
>     **case** ***LEADING***:
> 
>     ......

9）如果是LOOKING进行选举，则分为三种情况：

第一种：接收的选票小于自己（什么都不做，放弃投票）

第二种：接收的选票大于自己的（如果投票中的服务器比直接的那台服务器更优秀，则更新增加投票并发送投票）。

第三种：接收的选票等于自己的，如果此投票中的服务器比直接的那台服务器更优秀，则更新增加投票并发送投票。

a）然后保存信息到recvset

b）termPredicate方法自己投票的那台 服务器不能当领导（过半机器）

c）如果选出了Leader，那么还要判断在这个时间段内是否又来了新的投票，很可能需要重新选举，如果新的投票比选举出开的服务器更优秀，则重新选举，知道没有获取新的结果。

d）最后比较最终的结果是不是自己，如果是则更新为leader不是则更新为follower。

case LOOKING:
    
       // If notification > current, replace and send messages out
        if (n.electionEpoch > logicalclock.get()) {
            logicalclock.set(n.electionEpoch);
            recvset.clear();
            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()) {
            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.get(), proposedEpoch))) {
            // Verify if there is any change in the proposed leader
            while((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null){
                if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,proposedLeader, proposedZxid, proposedEpoch)){
                    recvqueue.put(n);
                    break;
                }
            }
            // This predicate is true once we don't read any new elevant message from the reception queue
            if (n == null) {
                self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING: learningState());
                Vote endVote = new Vote(proposedLeader,proposedZxid,logicalclock.get(),proposedEpoch);
                leaveInstance(endVote);
                return endVote;
            }
        }
        break;
    case OBSERVING:

## **二、Zookeeper使用注意事项** ##

### **1、Zk数据与日志清理** ###

dataDir目录、dataLogDir两个目录会随着时间推移变得庞大，容易造成硬盘满了，清理办法：

·自己编写shell脚本，保留最新的n个文件。

·使用zk自带的zkClient.sh保留最新的n个文件，zkClient.sh –n 1。

·配置autopurge.snapRetainCount和autopurge.purgeInterval两个参数配合使用。

### **2、Too many connections** ###

配置maxClientCnxns参数，配置单个客户端机器创建的最大连接数；

### **3、磁盘管理** ###

磁盘的I/O性能直接制约zookeeper更新操作速度，为了提高zk的写性能建议：

使用单独的磁盘

Jvm堆内存设置要小心

### **4、磁盘管理集群数量** ###

集群中机器的数量并不是越多越好，一个写操作需要半数以上的节点ack，所以集群节点数越多，整个集群可以抗挂点的节点数越多(越可靠)，但是吞吐量越差。集群的数量必须为奇数；

zk是基于内存进行读写操作的，有时候会进行消息广播，因此不建议在节点存取容量比较大的数据；

## **三、常见面试题** ##

### **1、ZAB协议是什么？** ###

ZAB协议是为分布式协调服务Zookeeper专门设计的一种支持崩溃恢复的原子广播协议，ZAB协议包括两种基本的模式：崩溃恢复和消息广播。

当整个zookeeper集群刚刚启动或者Leader服务器宕机、重启或者网络故障导致不存在过半的服务器与Leader服务器保持正常通信时，所有进程（服务器）进入崩溃恢复模式，首先选举产生新的Leader服务器，然后集群中Follower服务器开始与新的Leader服务器进行数据同步，当集群中超过半数机器与该Leader服务器完成数据同步之后，退出恢复模式进入消息广播模式，Leader服务器开始接收客户端的事务请求生成事物提案来进行事务请求处理。

### **2、Znode有哪几种类型** ###

PERSISTENT-持久节点：

除非手动删除，否则节点一直存在于Zookeeper上

EPHEMERAL-临时节点：

临时节点的生命周期与客户端会话绑定，一旦客户端会话失效（客户端与zookeeper连接断开不一定会话失效），那么这个客户端创建的所有临时节点都会被移除。

PERSISTENT\_SEQUENTIAL-持久顺序节点：

基本特性同持久节点，只是增加了顺序属性，节点名后边会追加一个由父节点维护的自增整型数字。

EPHEMERAL\_SEQUENTIAL-临时顺序节点：

基本特性同临时节点，增加了顺序属性，节点名后边会追加一个由父节点维护的自增整型数字。

### **3、ACL权限控制机制** ###

UGO（User/Group/Others）：

目前在Linux/Unix文件系统中使用，也是使用最广泛的权限控制方式。是一种粗粒度的文件系统权限控制模式。

ACL（Access Control List）访问控制列表，包括三个方面：

·权限模式（Scheme）

IP：从IP地址粒度进行权限控制

Digest：最常用，用类似于 username:password 的权限标识来进行权限配置，便于区分不同应用来进行权限控制

World：最开放的权限控制方式，是一种特殊的digest模式，只有一个权限标识“world:anyone”

Super：超级用户

·授权对象

授权对象指的是权限赋予的用户或一个指定实体，例如IP地址或是机器灯。

·权限 Permission

CREATE：数据节点创建权限，允许授权对象在该Znode下创建子节点

DELETE：子节点删除权限，允许授权对象删除该数据节点的子节点

READ：数据节点的读取权限，允许授权对象访问该数据节点并读取其数据内容或子节点列表等

WRITE：数据节点更新权限，允许授权对象对该数据节点进行更新操作

ADMIN：数据节点管理权限，允许授权对象对该数据节点进行ACL相关设置操作

### **4、ZK的角色有哪些** ###

**Leader**

事务请求的唯一调度和处理者，保证集群事务处理的顺序性。

集群内部各服务的调度者。

**Follower**

处理客户端的非事务请求，转发事务请求给Leader服务器。

参与事务请求Proposal的投票。

参与Leader选举投票。

**Observer**

·3.3.0版本以后引入的一个服务器角色，在不影响集群事务处理能力的基础上提升集群的非事务处理能力。

·处理客户端的非事务请求，转发事务请求给Leader服务器。

·不参与任何形式的投票。

### **5、ZK Server的工作状态有哪些，能否具体描述** ###

服务器具有四种状态，分别是LOOKING、FOLLOWING、LEADING、OBSERVING。

LOOKING：寻找Leader状态。当服务器处于该状态时，它会认为当前集群中没有Leader，因此需要进入Leader选举状态。

FOLLOWING：跟随者状态。表明当前服务器角色是Follower。

LEADING：领导者状态。表明当前服务器角色是Leader。

OBSERVING：观察者状态。表明当前服务器角色是Observer。

### **6、ZK的watch机制是否永久** ###

否。

### **7、ZK的常见客户端有哪些** ###

java客户端：zk自带的zkclient及Apache开源的Curator

### **8、分布式锁用zk怎么实现** ###

有两种实现方式，具体看课程代码

### **9、Zk默认的通信框架是什么** ###

Nio,可以改成netty

### **10、消息广播的流程** ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 12][]

*  Leader 接收到消息请求后，将消息赋予一个全局唯一的 64 位自增 id，叫做：zxid，通过 zxid 的大小比较即可实现因果有序这一特性。
 *   Leader 通过先进先出队列(通过 TCP 协议来实现，以此实现了全局有序这一特性)将带有 zxid 的消息作为一个提案(proposal)分发给所有 follower。
 *  当 follower 接收到 proposal，先将 proposal 写到硬盘，写硬盘成功后再向 leader 回一个 ACK。
 *  当 leader 接收到合法数量的 ACKs 后，leader 就向所有 follower 发送 COMMIT 命令，会在本地执行该消息。
 *  当 follower 收到消息的 COMMIT 命令时，就会执行该消息

### **11、领导选举的流程** ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 13][]

*  每个Server会发出一个投票,第一次都是投自己。投票信息：（myid，ZXID）
 *  收集来自各个服务器的投票
 *  处理投票并重新投票，处理逻辑：优先比较ZXID,然后比较myid
 *  统计投票，只要超过半数的机器接收到同样的投票信息，就可以确定leader
 *  改变服务器状态

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/2a25d2dd905b4034bee438a573613953.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/d688965584b2436085c2ba09c85e1570.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/b872eadd0f714d7a8242b71900e5c4e2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/77b0e282e0414deabe9a822825e245ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/f6c0af1d266941e597adcd2389c0d9f3.png
[20190825234235666.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/891fc5049d64493fa89047d4a908d2cf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/5214f955391d4bfda00160331bcea798.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/c48a87b7baac491288ee2d61646f6e8b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 7]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/436a23250a1440da963dbc2f844901ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 8]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/182e3635eb52424cb3ff7aca49662ecc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 9]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/1df23c27be6a4e04895ca7e8ddf292bc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 10]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/2d637794103f4d8081f5a0961680dacc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 11]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/378cd400f1a5408d8052c7b5fd37c560.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 12]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/8ed42ea06c1c47b7b343785728561a00.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 13]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/37eab130d96d4d659e378ceb183586e1.png