解读源码系列——源码级图文详解Zookeeper选举机制

public synchronized void start() {    if (!getView().containsKey(myid)) {        throw new RuntimeException("My id" + myid + " not in the peer list");    }       // 冷启动数据恢复    loadDataBase();    startServerCnxnFactory();    try {              // 启动通信工厂实例对象        adminServer.start();    } catch (AdminServerException e) {        LOG.warn("Problem startingAdminServer", e);        System.out.println(e);    }       // 准备选举环境    startLeaderElection();       // 执行选举    super.start();}

public void run() {    updateThreadName();     LOG.debug("Starting quorumpeer");    try {        jmxQuorumBean = new QuorumBean(this);       MBeanRegistry.getInstance().register(jmxQuorumBean, null);        for(QuorumServer s:getView().values()){            ZKMBeanInfo p;            if (getId() == s.id) {                p = jmxLocalPeerBean = new LocalPeerBean(this);                try {                   MBeanRegistry.getInstance().register(p, jmxQuorumBean);                } catch (Exception e) {                    LOG.warn("Failed to register withJMX", e);                    jmxLocalPeerBean = null;                }            } else {                RemotePeerBean rBean = new RemotePeerBean(this, s);                try {                    MBeanRegistry.getInstance().register(rBean, jmxQuorumBean);                    jmxRemotePeerBean.put(s.id, rBean);                } catch (Exception e) {                    LOG.warn("Failed toregister with JMX", e);                }            }        }    } catch (Exception e) {        LOG.warn("Failed to register withJMX", e);        jmxQuorumBean = null;    }     try {        /*         * Main loop         */        while (running) {            switch (getPeerState()) {            case LOOKING:                LOG.info("LOOKING");                 if(Boolean.getBoolean("readonlymode.enabled")) {                    LOG.info("Attemptingto start ReadOnlyZooKeeperServer");                     // Create read-only serverbut don't start it immediately                    final ReadOnlyZooKeeperServer roZk =                        new ReadOnlyZooKeeperServer(logFactory, this, this.zkDb);                     // Instead of starting roZkimmediately, wait some grace                    // period before we decidewe're partitioned.                    //                    // Thread is used herebecause otherwise it would require                    // changes in each ofelection strategy classes which is                    // unnecessary codecoupling.                    Thread roZkMgr = newThread() {                        public void run() {                            try {                                // lower-boundgrace 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 startReadOnlyZooKeeperServer, not started");                            } catch (Exceptione) {                               LOG.error("FAILED to start ReadOnlyZooKeeperServer", e);                            }                        }                    };                    try {                        roZkMgr.start();                        reconfigFlagClear();                        if (shuttingDownLE) {                            shuttingDownLE =false;                           startLeaderElection();                        }                        // 进行选举，选举结束，返回最终成为Leader胜选的那张选票                        setCurrentVote(makeLEStrategy().lookForLeader());                    } catch (Exception e) {                       LOG.warn("Unexpected exception", e);                       setPeerState(ServerState.LOOKING);                    } finally {                        // If the thread is inthe the grace period, interrupt                        // to come out ofwaiting.                        roZkMgr.interrupt();                        roZk.shutdown();                    }                } else {                    try {                       reconfigFlagClear();                        if (shuttingDownLE) {                           shuttingDownLE =false;                          startLeaderElection();                           }                       setCurrentVote(makeLEStrategy().lookForLeader());                    } catch (Exception e) {                       LOG.warn("Unexpected exception", e);                       setPeerState(ServerState.LOOKING);                    }                                       }                break;            case OBSERVING:                try {                   LOG.info("OBSERVING");                    setObserver(makeObserver(logFactory));                    observer.observeLeader();                } catch (Exception e) {                    LOG.warn("Unexpectedexception",e );                } finally {                    observer.shutdown();                    setObserver(null);                    updateServerState();                }                break;            case FOLLOWING:                try {                  LOG.info("FOLLOWING");                    setFollower(makeFollower(logFactory));                    follower.followLeader();                } catch (Exception e) {                   LOG.warn("Unexpectedexception",e);                } finally {                   follower.shutdown();                   setFollower(null);                   updateServerState();                }                break;            case LEADING:                LOG.info("LEADING");                try {                    setLeader(makeLeader(logFactory));                    leader.lead();                    setLeader(null);                } catch (Exception e) {                    LOG.warn("Unexpectedexception",e);                } finally {                    if (leader != null) {                       leader.shutdown("Forcing shutdown");                        setLeader(null);                    }                    updateServerState();                }                break;            }            start_fle =Time.currentElapsedTime();        }    } finally {        ... ...    }}

public Vote lookForLeader()throws InterruptedException {    try {        self.jmxLeaderElectionBean = newLeaderElectionBean();        MBeanRegistry.getInstance().register(                self.jmxLeaderElectionBean,self.jmxLocalPeerBean);    } catch (Exception e) {        LOG.warn("Failed to register withJMX", e);        self.jmxLeaderElectionBean = null;    }    if (self.start_fle == 0) {       self.start_fle =Time.currentElapsedTime();    }    try {              // 正常启动中，所有其他服务器，都会给我发送一个投票              // 保存每一个服务器的最新合法有效的投票        HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();                           // 存储合法选举之外的投票结果        HashMap<Long, Vote> outofelection= new HashMap<Long, Vote>();               // 一次选举的最大等待时间，默认值是0.2s        int notTimeout = finalizeWait;                           // 每发起一轮选举，logicalclock++              // 在没有合法的epoch数据之前，都使用逻辑时钟代替              // 选举leader的规则：依次比较 epoch（任期）  zxid（事务id）  serverid（myid）  谁大谁当选leader        synchronized(this){                     // 更新逻辑时钟，每进行一次选举，都需要更新逻辑时钟                     // logicalclock = epoch            logicalclock.incrementAndGet();                     // 更新选票（serverid，zxid, epoch），            updateProposal(getInitId(), getInitLastLoggedZxid(),getPeerEpoch());        }         LOG.info("New election. My id=  " + self.getId() +                ", proposed zxid=0x"+ Long.toHexString(proposedZxid));              // 广播选票，把自己的选票发给其他服务器        sendNotifications();         /*         * Loop in which we exchangenotifications until we find a leader         */              // 一轮一轮的选举直到选举成功        while ((self.getPeerState() ==ServerState.LOOKING) &&                (!stop)){            … …        }        return null;    } finally {        … …    }}

private void sendNotifications() {       // 遍历投票参与者，给每台服务器发送选票    for (long sid :self.getCurrentAndNextConfigVoters()) {        QuorumVerifier qv =self.getQuorumVerifier();              // 创建发送选票        ToSend notmsg = new ToSend(ToSend.mType.notification,                proposedLeader,                proposedZxid,                logicalclock.get(),                QuorumPeer.ServerState.LOOKING,                sid,                proposedEpoch,qv.toString().getBytes());                                   if(LOG.isDebugEnabled()){            LOG.debug("SendingNotification: " + proposedLeader + " (n.leader), 0x"  +                 Long.toHexString(proposedZxid) + " (n.zxid), 0x" +Long.toHexString(logicalclock.get())  +                  " (n.round), " +sid + " (recipient), " + self.getId() +                  " (myid), 0x" +Long.toHexString(proposedEpoch) + " (n.peerEpoch)");        }              // 把发送选票放入发送队列        sendqueue.offer(notmsg);    }}

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;            }        }        LOG.info("WorkerSender isdown");    }     /**     * Called by run() once there is a newmessage to send.     * @param m     message to send     */    void process(ToSend m) {        ByteBuffer requestBuffer =buildMsg(m.state.ordinal(),                                           m.leader,                                           m.zxid,                                           m.electionEpoch,                                            m.peerEpoch,                                           m.configData);              // 发送选票        manager.toSend(m.sid, requestBuffer);    }}  public void toSend(Long sid,ByteBuffer b) {    /*     * If sending message to myself, thensimply enqueue it (loopback).     */       // 判断如果是发给自己的消息，直接进入自己的RecvQueue    if (this.mySid == sid) {         b.position(0);         addToRecvQueue(new Message(b.duplicate(), sid));        /*         * Otherwise send to the correspondingthread to send.         */    } else {         /*          * Start a new connection if doesn'thave one already.          */               // 如果是发给其他服务器，创建对应的发送队列或者获取已经存在的发送队列               // ，并把要发送的消息放入该队列         ArrayBlockingQueue<ByteBuffer>bq = new ArrayBlockingQueue<ByteBuffer>(            SEND_CAPACITY);         ArrayBlockingQueue<ByteBuffer>oldq = queueSendMap.putIfAbsent(sid, bq);         if (oldq != null) {             addToSendQueue(oldq, b);         } else {             addToSendQueue(bq, b);         }               // 将选票发送出去         connectOne(sid);    }}

public void addToRecvQueue(Messagemsg) {    synchronized(recvQLock) {        if (recvQueue.remainingCapacity() == 0){            try {                recvQueue.remove();            } catch (NoSuchElementException ne){                // element could be removed bypoll()                 LOG.debug("Trying toremove from an empty " +                     "recvQueue. Ignoringexception " + ne);            }        }        try {                     // 将发送给自己的选票添加到recvQueue队列            recvQueue.add(msg);        } catch (IllegalStateException ie) {            // This should never happen            LOG.error("Unable to insert elementin the recvQueue " + ie);        }    }}    

private void addToSendQueue(ArrayBlockingQueue<ByteBuffer>queue,     ByteBuffer buffer) {    if(queue.remainingCapacity() == 0) {        try {           queue.remove();        } catch(NoSuchElementException ne) {            //element could be removed by poll()           LOG.debug("Trying to remove from an empty " +                   "Queue. Ignoring exception " + ne);        }    }    try {              //将要发送的消息添加到发送队列        queue.add(buffer);    } catch(IllegalStateException ie) {        // Thisshould never happen       LOG.error("Unable to insert an element in the queue " + ie);    }}

synchronized void connectOne(long sid){    if(senderWorkerMap.get(sid) != null) {       LOG.debug("There is a connection already for server " + sid);        return;    }   synchronized (self.QV_LOCK) {        boolean knownId = false;        //Resolve hostname for the remote server before attempting to        //connect in case the underlying ip address has changed.       self.recreateSocketAddresses(sid);       Map<Long, QuorumPeer.QuorumServer> lastCommittedView =self.getView();       QuorumVerifier lastSeenQV = self.getLastSeenQuorumVerifier();       Map<Long, QuorumPeer.QuorumServer> lastProposedView =lastSeenQV.getAllMembers();                     if(lastCommittedView.containsKey(sid)) {           knownId = true;            if(connectOne(sid,lastCommittedView.get(sid).electionAddr))               return;        }        if(lastSeenQV != null && lastProposedView.containsKey(sid)               && (!knownId || (lastProposedView.get(sid).electionAddr !=               lastCommittedView.get(sid).electionAddr))) {           knownId = true;            if(connectOne(sid,lastProposedView.get(sid).electionAddr))               return;        }        if(!knownId) {           LOG.warn("Invalid server id: " + sid);           return;        }    }}  synchronized private boolean connectOne(long sid, InetSocketAddresselectionAddr){    if(senderWorkerMap.get(sid) != null) {       LOG.debug("There is a connection already for server " + sid);        returntrue;    }     Socket sock= null;    try {       LOG.debug("Opening channel to server " + sid);        if(self.isSslQuorum()) {            SSLSocket sslSock = self.getX509Util().createSSLSocket();            setSockOpts(sslSock);            sslSock.connect(electionAddr, cnxTO);            sslSock.startHandshake();            sock = sslSock;            LOG.info("SSL handshake complete with {} - {} - {}",sslSock.getRemoteSocketAddress(), sslSock.getSession().getProtocol(),sslSock.getSession().getCipherSuite());         } else{            sock = new Socket();            setSockOpts(sock);             sock.connect(electionAddr, cnxTO);          }        LOG.debug("Connected to server " + sid);        //Sends connection request asynchronously if the quorum        // saslauthentication is enabled. This is required because        // saslserver authentication process may take few seconds to        //finish, this may delay next peer connection requests.        if(quorumSaslAuthEnabled) {            initiateConnectionAsync(sock, sid);        } else{                     //处理连接            initiateConnection(sock,sid);        }        returntrue;    } catch(UnresolvedAddressException e) {        ... ...    }} public void initiateConnection(final Socket sock, final Long sid) {    try {        startConnection(sock,sid);    } catch(IOException e) {       LOG.error("Exception while connecting, id: {}, addr: {}, closinglearner connection",               new Object[] { sid, sock.getRemoteSocketAddress() }, e);       closeSocket(sock);        return;    }}

private boolean startConnection(Socket sock, Long sid)        throws IOException {   DataOutputStream dout = null;   DataInputStream din = null;    try {        // UseBufferedOutputStream to reduce the number of IP packets. This is        //important for x-DC scenarios.              //通过输出流，向服务器发送数据       BufferedOutputStream buf = new BufferedOutputStream(sock.getOutputStream());        dout = newDataOutputStream(buf);         //Sending id and challenge        //represents protocol version (in other words - message type)       dout.writeLong(PROTOCOL_VERSION);       dout.writeLong(self.getId());        String addr = formatInetAddr(self.getElectionAddress());        byte[] addr_bytes = addr.getBytes();       dout.writeInt(addr_bytes.length);       dout.write(addr_bytes);       dout.flush();              //通过输入流读取对方发送过来的选票        din = new DataInputStream(                newBufferedInputStream(sock.getInputStream()));    } catch(IOException e) {       LOG.warn("Ignoring exception reading or writing challenge: ",e);       closeSocket(sock);        returnfalse;    }     // authenticatelearner   QuorumPeer.QuorumServer qps = self.getVotingView().get(sid);    if (qps !=null) {        // TODO- investigate why reconfig makes qps null.       authLearner.authenticate(sock, qps.hostname);    }     // If lostthe challenge, then drop the new connection           //如果对方的id比我的大，我是没有资格给对方发送连接请求的，直接关闭自己的客户端    if (sid > self.getId()) {       LOG.info("Have smaller server identifier, so dropping the " +               "connection: (" + sid + ", " + self.getId() +")");       closeSocket(sock);        //Otherwise proceed with the connection    } else {              //初始化，发送器和接收器        SendWorker sw = new SendWorker(sock,sid);        RecvWorker rw = new RecvWorker(sock,din, sid, sw);       sw.setRecv(rw);        SendWorker vsw = senderWorkerMap.get(sid);         if(vsw!= null)           vsw.finish();                    senderWorkerMap.put(sid, sw);       queueSendMap.putIfAbsent(sid, new ArrayBlockingQueue<ByteBuffer>(               SEND_CAPACITY));                                         //启动发送器线程和接收器线程        sw.start();        rw.start();         return true;    }    return false;}

public void run() {   thread Cnt.incrementAndGet();    try {        /**         * If there is nothing in the queue to send, then we         * send the lastMessage to ensure that the last message         * was received by the peer. The message could be dropped         * incase self or the peer shutdown their connection         * (and exit the thread) prior toreading/processing         * the last message. Duplicate messages are handled correctly         * by the peer.         *         * If the send queue is non-empty, then we have a recent         *message than that stored in lastMessage. To avoid sending         *stale message, we should send the message in the send queue.         */       Array BlockingQueue<ByteBuffer> bq = queueSendMap.get(sid);        if (bq== null || isSendQueueEmpty(bq)) {          ByteBuffer b = lastMessageSent.get(sid);           if(b != null) {              LOG.debug("Attempting to send lastMessage to sid=" + sid);              send(b);           }        }    } catch(IOException e) {        LOG.error("Failedto send last message. Shutting down thread.", e);       this.finish();    }     try {              //只要连接没有断开        while(running && !shutdown && sock != null) {            ByteBuffer b = null;            try{               ArrayBlockingQueue<ByteBuffer> bq = queueSendMap                        .get(sid);               if (bq != null) {                                   //不断从发送队列SendQueue中，获取发送消息，并执行发送                   b = pollSendQueue(bq,1000, TimeUnit.MILLISECONDS);               } else {                   LOG.error("No queue of incoming messages for " +                              "server" + sid);                   break;               }                if(b != null){                                   //更新对于sid这台服务器的最近一条消息                    lastMessageSent.put(sid, b);                                                                     //执行发送                   send(b);               }            }catch (InterruptedException e) {               LOG.warn("Interrupted while waiting for message on queue",                        e);            }        }    } catch(Exception e) {       LOG.warn("Exception when using channel: for id " + sid                + " my id = " + QuorumCnxManager.this.mySid                + " error = " + e);    }   this.finish();   LOG.warn("Send worker leaving thread " + " id " +sid + " my id = " + self.getId());} synchronized void send(ByteBuffer b) throws IOException {    byte[] msgBytes = new byte[b.capacity()];    try {       b.position(0);       b.get(msgBytes);    } catch(BufferUnderflowException be) {       LOG.error("BufferUnderflowException ", be);        return;    }   // 输出流向外发送   dout.writeInt(b.capacity());    dout.write(b.array());   dout.flush();}

public void run() {   threadCnt.incrementAndGet();    try {              //只要连接没有断开        while(running && !shutdown && sock != null) {            /**             *Reads the first int to determine the length of the             *message             */            int length = din.readInt();            if(length <= 0 || length > PACKETMAXSIZE) {               throw new IOException(                        "Received packetwith invalid packet: "                               + length);            }            /**             *Allocates a new ByteBuffer to receive the message             */           byte[] msgArray = new byte[length];            //输入流接收消息            din.readFully(msgArray, 0, length);           ByteBuffer message = ByteBuffer.wrap(msgArray);                     //接收对方发送过来的选票            addToRecvQueue(newMessage(message.duplicate(), sid));        }    } catch(Exception e) {       LOG.warn("Connection broken for id " + sid + ", my id ="                + QuorumCnxManager.this.mySid + ", error = " , e);    } finally {       LOG.warn("Interrupting SendWorker");       sw.finish();       closeSocket(sock);    }} public void addToRecvQueue(Message msg) {   synchronized(recvQLock) {        if(recvQueue.remainingCapacity() == 0) {            try{               recvQueue.remove();            }catch (NoSuchElementException ne) {               // element could be removed by poll()                LOG.debug("Trying to remove from an empty " +                    "recvQueue. Ignoring exception " + ne);            }        }        try {                     //将接收到的消息，放入接收消息队列            recvQueue.add(msg);        } catch(IllegalStateException ie) {            //This should never happen           LOG.error("Unable to insert element in the recvQueue " + ie);        }    }}

class WorkerReceiver extends ZooKeeperThread  {    volatileboolean 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{                            //从RecvQueue中取出选举投票消息（其他服务器发送过来的）               response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);               … …            }catch (InterruptedException e) {               LOG.warn("Interrupted Exception while waiting for new message"+                        e.toString());            }        }       LOG.info("WorkerReceiver is down");    }}