zookeeper原理之选举流程分析

前面分析这么多，还没有正式分析到 leader 选举的核心流程，前期准备工作做好了以后，接下来就开始正式分析 leader 选举的过程：

public synchronized void start() {
     loadDataBase();
     cnxnFactory.start(); 
     startLeaderElection();
     super.start(); //启动线程
}

很明显，super.start() 表示当前类 QuorumPeer 继承了线程，线程必须要重写 run 方法，所以我们可以在 QuorumPeer 中找到一个 run 方法。

 

QuorumPeer.run

这段代码的逻辑比较长。粗略看一下结构，好像也不难

PeerState 有几种状态，分别是

1. LOOKING，竞选状态。
2. FOLLOWING，随从状态，同步 leader 状态，参与投票。
3. OBSERVING，观察状态,同步 leader 状态，不参与投票。
4. LEADING，领导者状态。

对于选举来说，默认都是 LOOKING 状态，只有 LOOKING 状态才会去执行选举算法。每个服务器在启动时都会选择自己做为领导，然后将投票信息发送出去，循环一直到选举出领导为止。

@Override
	public void run() {
		setName("QuorumPeer" + "[myid=" + getId() + "]" + cnxnFactory.getLocalAddress());
		// … 根据选举状态，选择不同的处理方式
		while (running) {
			switch (getPeerState()) {
			case LOOKING:
				LOG.info("LOOKING");
				// 判断是否为只读模式,通过”readonlymode.enabled”开 启
				if (Boolean.getBoolean("readonlymode.enabled")) {
					// 只读模式的启动流程
				} else {
					try {
						setBCVote(null);
						// 设置当前的投票，通过策略模式来决定当前用哪个选举算法来进行领导选举

						setCurrentVote(makeLEStrategy().lookForLeader());
					} catch (Exception e) {
						LOG.warn("Unexpected exception", e);
						setPeerState(ServerState.LOOKING);
					}
				}
				break;
			// …后续逻辑暂时不用管
			}
		}
	}

FastLeaderElection.lookForLeader

开始发起投票流程

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.incrementAndGet();//更新逻辑时钟，用来判断是否在同一轮选举周期
				//初始化选票数据：这里其实就是把当前节点的 myid，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 exchange notifications until we find a leader
			 */
			//这里就是不断循环，根据投票信息进行进行 leader 选举
			while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){
				/*
				 * Remove next notification from queue, timesout after 2 times
				 * the termination time
				 */
				//从 recvqueue 中获取消息
				Notification n = recvqueue.poll(notTimeout,TimeUnit.MILLISECONDS);
				/*
				 * Sends more notifications if haven't received enough.
				 * Otherwise processes new notification.
				 */
				if(n == null){ //如果没有获取到外部的投票，有可能是集群之间的节点没有真正连接上
						if(manager.haveDelivered()){//判断发送队列是否有数据，如果发送队列为空，再发一次自己的选票
							sendNotifications();
						} else {//在此发起集群节点之间的连接
							manager.connectAll();
						}
						/*
						 * Exponential backoff
						 */
						int tmpTimeOut = notTimeout*2;
						notTimeout = (tmpTimeOut < maxNotificationInterval? tmpTimeOut : maxNotificationInterval);
						LOG.info("Notification time out: " + notTimeout);
					}
				}
			}
	}

选票的判断逻辑（核心代码）

 

// 判断收到的选票中的 sid 和选举的 leader 的 sid 是否存在于我们集群所配置的 myid 范围
		if (validVoter(n.sid) && validVoter(n.leader)) {
			// 判断接收到的投票者的状态，默认是 LOOKING 状态,说明当前发起投票的服务器也是在找 leader
			switch (n.state) {
			case LOOKING: // 说明当前发起投票的服务器也是在找 leader
				// 如果收到的投票的逻辑时钟大于当前的节点的逻辑时钟
				if (n.electionEpoch > logicalclock.get()) {
					logicalclock.set(n.electionEpoch);// 更新成新一轮的逻辑时钟
					recvset.clear();
					// 比较接收到的投票和当前节点的信息进行比较，比较的顺序epoch、zxid、myid,如果返回
					// true，则更新当前节点的票据（sid,zxid,epoch）,那么下次再发起投票的时候，就不再是选自己了
					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()) {// 如果小于，说明收到的票据已经过期了，直接把这张票丢掉
					if (LOG.isDebugEnabled()) {
						LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
								+ Long.toHexString(n.electionEpoch) + ", logicalclock=0x"
								+ Long.toHexString(logicalclock.get()));
					}
					break; // 这个判断表示收到的票据的 epoch 是相同的，那么按照 epoch、zxid、myid
							// 顺序进行比较比较成功以后，把对方的票据信息更新到自己的节点
				} else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid,
						proposedEpoch)) {
					updateProposal(n.leader, n.zxid, n.peerEpoch);
					sendNotifications();// 把收到的票据再发出去告诉大家我要选 n.leader 为 leader
				}
				if (LOG.isDebugEnabled()) {
					LOG.debug("Adding vote: from=" + n.sid + ", proposed leader=" + n.leader + ", proposed zxid=0x"
							+ Long.toHexString(n.zxid) + ", proposed election epoch=0x"
							+ Long.toHexString(n.electionEpoch));
				}
				// 将收到的投票信息放入投票的集合 recvset 中, 用来作为最终的 "过半原则" 判断
				recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
				// 判断选举是否结束
				if (termPredicate(recvset, new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch))) {
					// 进入这个判断，说明选票达到了 leader 选举的要求
					// 在更新状态之前，服务器会等待 finalizeWait 毫秒时间来接收新的选票，以防止漏下关键选票如果收到可能改变
					// Leader 的新选票，则重新进行计票
					while ((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
						if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid,
								proposedEpoch)) {
							recvqueue.put(n);
							break;
						}
					}
					// 如果 notifaction 为空，说明 Leader 节点是可以确定好了
					if (n == null) {
						// 设置当前当前节点的状态（判断 leader 节点是不是我自己，如果是，直接更新当前节点的 state 为
						// LEADING）否则，根据当前节点的特性进行判断，决定是FOLLOWING 还是 OBSERVING
						self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING : learningState());
						// 组装生成这次 Leader 选举最终的投票的结果
						Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch);
						leaveInstance(endVote);// 清空
						return endVote; // 返回最终的票据
					}
				}
				break;
			case OBSERVING:// OBSERVING 不参与 leader 选举
				LOG.debug("Notification from observer: " + n.sid);
				break;
			case FOLLOWING:
			case LEADING:
				/*
				 * Consider all notifications from the same epoch together.
				 */
				if (n.electionEpoch == logicalclock.get()) {
					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.set(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;
			}
		}

投票处理的流程图

termPredicate

这个方法是使用过半原则来判断选举是否结束，如果返回 true，说明能够选出 leader 服务器votes 表示收到的外部选票的集合vote 表示但前服务器的选票。

protected boolean termPredicate(HashMap<Long, Vote> votes, Vote vote) {
		HashSet<Long> set = new HashSet<Long>();
		// 遍历接收到的所有选票数据
		for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
			// 对选票进行归纳，就是把所有选票数据中和当前节点的票据相同的票据进行统计
			if (vote.equals(entry.getValue())) {
				set.add(entry.getKey());
			}
		} // 对选票进行判断
		return self.getQuorumVerifier().containsQuorum(set);
	}

QuorumMaj. containsQuorum

判断当前节点的票数是否是大于一半，默认采用 QuorumMaj 来实现。

public boolean containsQuorum(Set<Long> set){
     return (set.size() > half);
}

这个 half 的值是多少呢？

可以在 QuorumPeerConfig.parseProperties 这个方法中，找到如下代码。

也就是说，在构建 QuorumMaj 的时候，传递了当前集群节点的数量，这里是 3那么，hafl=3/2=1

public QuorumMaj(int n){
     this.half = n/2; 
}

那么 set.size()>1. 意味着至少要有两个节点的票据是选择你当 leader，否则，还得继续投。