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

protected void initializeAndRun(String[] args)
    throws ConfigException, IOException
{
	// 加载配置的类
    QuorumPeerConfig config = new QuorumPeerConfig();
    if (args.length == 1) {
    	// 从配置文件加载配置到内存中
        config.parse(args[0]);
    }

    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 {
      ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
      cnxnFactory.configure(config.getClientPortAddress(),
                            config.getMaxClientCnxns());

	  // 将配置信息设置到QuorumPeer类
      quorumPeer = new QuorumPeer();
      quorumPeer.setClientPortAddress(config.getClientPortAddress());
      quorumPeer.setTxnFactory(new FileTxnSnapLog(
                  new File(config.getDataLogDir()),
                  new File(config.getDataDir())));
      quorumPeer.setQuorumPeers(config.getServers());
      quorumPeer.setElectionType(config.getElectionAlg());
      quorumPeer.setMyid(config.getServerId());
      quorumPeer.setTickTime(config.getTickTime());
      quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
      quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
      quorumPeer.setInitLimit(config.getInitLimit());
      quorumPeer.setSyncLimit(config.getSyncLimit());
      quorumPeer.setQuorumVerifier(config.getQuorumVerifier());
      quorumPeer.setCnxnFactory(cnxnFactory);
      quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
      quorumPeer.setLearnerType(config.getPeerType());
      quorumPeer.setSyncEnabled(config.getSyncEnabled());
      quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());

      quorumPeer.start(); // 开启子线程加载db信息以及开始选举
      quorumPeer.join(); // 主线程等待子线程完成
  } catch (InterruptedException e) {
      // warn, but generally this is ok
      LOG.warn("Quorum Peer interrupted", e);
  }
}

public synchronized void start() {
    loadDataBase(); // 这里是恢复DB信息，如epoch和ZXID等
    cnxnFactory.start();        
    startLeaderElection(); //开始选举的流程
    super.start(); // 启动线程
}

synchronized public void startLeaderElection() {
	try {
		// 投自己一票
		currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
	} catch(IOException e) {
		RuntimeException re = new RuntimeException(e.getMessage());
		re.setStackTrace(e.getStackTrace());
		throw re;
	}
	
	// 创建选举算法，默认是使用FastLeaderElection选举算法
    this.electionAlg = createElectionAlgorithm(electionType);
}

protected Election createElectionAlgorithm(int electionAlgorithm){
    Election le=null;
            
    //TODO: use a factory rather than a switch
    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 = new QuorumCnxManager(this);
        QuorumCnxManager.Listener listener = qcm.listener;
        if(listener != null){
            listener.start();
            // 默认会进入到这里，可以在zoo.cfg配置文件中配置
            le = new FastLeaderElection(this, qcm);
        } else {
            LOG.error("Null listener when initializing cnx manager");
        }
        break;
    default:
        assert false;
    }
    return le;
}

public void run() {
    try {
        while (running) {
        	// 根据服务器状态进入相应的流程，因为是选举流程，所以都是LOOKING状态，其它的流程可以暂时忽略
            switch (getPeerState()) {
            case LOOKING:
                if (Boolean.getBoolean("readonlymode.enabled")) { // 当前服务器为只读服务器，与我们的流程无关
                    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");
                            } catch (Exception e) {
                                LOG.error("FAILED to start ReadOnlyZooKeeperServer", e);
                            }
                        }
                    };
                    try {
                        roZkMgr.start();
                        setBCVote(null);
                        setCurrentVote(makeLEStrategy().lookForLeader());
                    } catch (Exception e) {
                        LOG.warn("Unexpected exception",e);
                        setPeerState(ServerState.LOOKING);
                    } finally {
                        // If the thread is in the the grace period, interrupt
                        // to come out of waiting.
                        roZkMgr.interrupt();
                        roZk.shutdown();
                    }
                } else {
                    try {
                        setBCVote(null);
                        // 选出leader
                        setCurrentVote(makeLEStrategy().lookForLeader());
                    } catch (Exception e) {
                        LOG.warn("Unexpected exception", e);
                        setPeerState(ServerState.LOOKING);
                    }
                }
                break;
            }
        }
    } finally {
        LOG.warn("QuorumPeer main thread exited");
        try {
            MBeanRegistry.getInstance().unregisterAll();
        } catch (Exception e) {
            LOG.warn("Failed to unregister with JMX", e);
        }
        jmxQuorumBean = null;
        jmxLocalPeerBean = null;
    }
}

// 发送和接收选票队列
LinkedBlockingQueue<ToSend> sendqueue;
LinkedBlockingQueue<Notification> recvqueue;

public Vote lookForLeader() throws InterruptedException {
   try {
   		// 存放收到的投票信息
       HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();
		// 存放当前服务器的投票信息
       HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();

       int notTimeout = finalizeWait;

		// 第一次投票都投自己
       synchronized(this){
           logicalclock++;
           updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
       }
		
		// 发送自己的投票信息
       sendNotifications();

		// 循环直到选出Leader
       while ((self.getPeerState() == ServerState.LOOKING) &&
               (!stop)){
           
           // 从队列中按顺序取出投票
           Notification n = recvqueue.poll(notTimeout,
                   TimeUnit.MILLISECONDS);

			
           if(n == null){ // 未收到任何投票信息
               if(manager.haveDelivered()){ // 发送队列空闲情况下，就继续发送自己的投票信息
                   sendNotifications();
               } else { // 发送队列不为null，可能是其它服务器还未启动，尝试重连
                   manager.connectAll();
               }

           }
           else if(self.getVotingView().containsKey(n.sid)) { // 该消息是否属于当前集群
               switch (n.state) { // 判断收到消息的节点的状态
               case LOOKING:
                   // 判断是否是新一轮的选举
                   if (n.electionEpoch > logicalclock) {
                       logicalclock = n.electionEpoch; // 更新logicalclock 
                       recvset.clear(); // 清空前一轮收到的投票信息
                       // 比较epoch、myid和zxid，并更新自己的投票为胜出的节点
                       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)) {
                       // 同一轮投票直接比较myid和zxid
                       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))) {

                       // 等待所有notification都被处理完，直到超时
                       while((n = recvqueue.poll(finalizeWait,
                               TimeUnit.MILLISECONDS)) != null){
                           if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                   proposedLeader, proposedZxid, proposedEpoch)){
                               recvqueue.put(n);
                               break;
                           }
                       }

                       // leader已经确定
                       if (n == null) {
                       		// 修改状态为leader或者follower
                           self.setPeerState((proposedLeader == self.getId()) ?
                                   ServerState.LEADING: learningState());

                           Vote endVote = new Vote(proposedLeader,
                                                   proposedZxid,
                                                   logicalclock,
                                                   proposedEpoch);
                           leaveInstance(endVote);
                           return endVote;
                       }
                   }
                   break;
               case OBSERVING: // observer不参与投票
                   LOG.debug("Notification from observer: " + n.sid);
                   break;
               // follower和leader都要参与投票
               case FOLLOWING:
               case LEADING:
                   /*
                    * Consider all notifications from the same epoch
                    * together.
                    */
                   if(n.electionEpoch == logicalclock){
                       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;
                       }
                   }

                   /*
                    * Before joining an established ensemble, verify
                    * a majority is following the same 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()) ?
                                   ServerState.LEADING: learningState());
                       }
                       Vote endVote = new Vote(n.leader,
                                               n.zxid,
                                               n.electionEpoch,
                                               n.peerEpoch);
                       leaveInstance(endVote);
                       return endVote;
                   }
                   break;
               default:
                   LOG.warn("Notification state unrecognized: {} (n.state), {} (n.sid)",
                           n.state, n.sid);
                   break;
               }
           } else {
               LOG.warn("Ignoring notification from non-cluster member " + n.sid);
           }
       }
       return null;
   } finally {
       try {
           if(self.jmxLeaderElectionBean != null){
               MBeanRegistry.getInstance().unregister(
                       self.jmxLeaderElectionBean);
           }
       } catch (Exception e) {
           LOG.warn("Failed to unregister with JMX", e);
       }
       self.jmxLeaderElectionBean = null;
   }
}

private void sendNotifications() {
    for (QuorumServer server : self.getVotingView().values()) {
        long sid = server.id;

        ToSend notmsg = new ToSend(ToSend.mType.notification,
                proposedLeader,
                proposedZxid,
                logicalclock,
                QuorumPeer.ServerState.LOOKING,
                sid,
                proposedEpoch);
        if(LOG.isDebugEnabled()){
            LOG.debug("Sending Notification: " + proposedLeader + " (n.leader), 0x"  +
                  Long.toHexString(proposedZxid) + " (n.zxid), 0x" + Long.toHexString(logicalclock)  +
                  " (n.round), " + sid + " (recipient), " + self.getId() +
                  " (myid), 0x" + Long.toHexString(proposedEpoch) + " (n.peerEpoch)");
        }
        sendqueue.offer(notmsg);
    }
}

public void run() {
    while (!stop) {
        try {
            ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
            if(m == null) continue;

            process(m); // 发送ToSend
        } catch (InterruptedException e) {
            break;
        }
    }
}

public void toSend(Long sid, ByteBuffer b) {
    if (self.getId() == sid) { // 发送给自己不需要走网络通信，直接放到接收队列中即可
         b.position(0);
         addToRecvQueue(new Message(b.duplicate(), sid));
        
    } else {
    	// 发送给其它节点，需要判断之前是否发送过
         if (!queueSendMap.containsKey(sid)) { 
         	// 未发送过则新建发送队列，SEND_CAPACITY为1，表示每次只发送一个消息
             ArrayBlockingQueue<ByteBuffer> bq = new ArrayBlockingQueue<ByteBuffer>(
                     SEND_CAPACITY);
             queueSendMap.put(sid, bq);
             addToSendQueue(bq, b);

         } else {
         	// 前一个消息还未发送完成，则重新发送
             ArrayBlockingQueue<ByteBuffer> bq = queueSendMap.get(sid);
             if(bq != null){
                 addToSendQueue(bq, b);
             } else {
                 LOG.error("No queue for server " + sid);
             }
         }
         connectOne(sid); // 真正的底层传输逻辑
            
    }
}