Zookeeper源码部分 第2章 2.7 服务端Follower启动

2.7 服务端Follower启动

FollowerZooKeeperServer

Ctrl + n全局查找Follower，然后ctrl + f 查找followleader(

void followleader() throws InterruptedException {
    self.end_fle = Time.currentelapsedtime();
    long electionTiMetaken = self.end_fle - self.start_fle;
    self.setElectionTiMetaken(electionTiMetaken);
    LOG.info("FOLLOWING - leader ELECTION TOOK - {} {}", electionTiMetaken,
            QuorumPeer.FLE_TIME_UNIT);
    self.start_fle = 0;
    self.end_fle = 0;
    fzk.registerJMX(new FollowerBean(this, zk), self.jmxLocalPeerBean);
    try {
        QuorumServer leaderServer = findleader();            
        try {
            connectToleader(leaderServer.addr, leaderServer.hostname);
            long newEpochZxid = registerWithleader(leader.FOLLOWERINFO);
            if (self.isReconfigStateChange())
               throw new Exception("learned about role change");
            //check to see if the leader zxid is lower than ours
            //this should never happen but is just a safety check
            long newEpoch = ZxidUtils.getEpochFromZxid(newEpochZxid);
            if (newEpoch < self.getAcceptedEpoch()) {
                LOG.error("Proposed leader epoch " + ZxidUtils.zxidToString(newEpochZxid)
                        + " is less than our accepted epoch " + ZxidUtils.zxidToString(self.getAcceptedEpoch()));
                throw new IOException("Error: Epoch of leader is lower");
            }
            syncWithleader(newEpochZxid);                
            QuorumPacket qp = new QuorumPacket();
            while (this.isRunning()) {
                readPacket(qp);
                processpacket(qp);
            }
        } catch (Exception e) {
            ... ...
        }
    } finally {
        zk.unregisterJMX((Learner)this);
    }
}

void readPacket(QuorumPacket pp) throws IOException {
    synchronized (leaderIs) {
        leaderIs.readRecord(pp, "packet");
    }
    if (LOG.isTraceEnabled()) {
        final long traceMask =
            (pp.getType() == leader.PING) ? ZooTrace.SERVER_PING_TRACE_MASK
                : ZooTrace.SERVER_PACKET_TRACE_MASK;

        ZooTrace.logQuorumPacket(LOG, traceMask, 'i', pp);
    }
}

protected void processpacket(QuorumPacket qp) throws Exception{
    switch (qp.getType()) {
    case leader.PING:            
        ping(qp);            
        break;
    case leader.PROPOSAL:           
        TxnHeader hdr = new TxnHeader();
        Record txn = SerializeUtils.deserializeTxn(qp.getData(), hdr);
        if (hdr.getZxid() != lastQueued + 1) {
            LOG.warn("Got zxid 0x"
                    + Long.toHexString(hdr.getZxid())
                    + " expected 0x"
                    + Long.toHexString(lastQueued + 1));
        }
        lastQueued = hdr.getZxid();
        
        if (hdr.getType() == OpCode.reconfig){
           SetDataTxn setDataTxn = (SetDataTxn) txn;       
           QuorumVerifier qv = self.configFromString(new String(setDataTxn.getData()));
           self.setLastSeenQuorumVerifier(qv, true);                               
        }
        
        fzk.logRequest(hdr, txn);
        break;
    case leader.COMMIT:
        fzk.commit(qp.getZxid());
        break;
        
    case leader.COMMITANDACTIVATE:
       // get the new configuration from the request
       Request request = fzk.pendingTxns.element();
       SetDataTxn setDataTxn = (SetDataTxn) request.getTxn();                                                                                                      
       QuorumVerifier qv = self.configFromString(new String(setDataTxn.getData()));                                

       // get new designated leader from (current) leader's message
       ByteBuffer buffer = ByteBuffer.wrap(qp.getData());    
       long suggestedleaderId = buffer.getLong();
        boolean majorChange = 
               self.processReconfig(qv, suggestedleaderId, qp.getZxid(), true);
       // commit (writes the new config to ZK tree (/zookeeper/config)                     
       fzk.commit(qp.getZxid());
        if (majorChange) {
           throw new Exception("changes proposed in reconfig");
       }
       break;
    case leader.UPTODATE:
        LOG.error("Received an UPTODATE message after Follower started");
        break;
    case leader.REVALIDATE:
        revalidate(qp);
        break;
    case leader.SYNC:
        fzk.sync();
        break;
    default:
        LOG.warn("UnkNown packet type: {}", LearnerHandler.packetToString(qp));
        break;
    }
}