首页 > 应用开发 > IT综合 > 搞懂分布式技术4:ZAB协议概述与选主流程详解
public void run() { //....... try { while (running) { switch (getPeerState()) { case LOOKING: if (Boolean.getBoolean("readonlymode.enabled")) { //... try { //投票给自己... setCurrentVote(makeLEStrategy().lookForLeader()); } catch (Exception e) { //... } finally { //... } } else { try { //... setCurrentVote(makeLEStrategy().lookForLeader()); } catch (Exception e) { //... } } break; case OBSERVING: //... break; case FOLLOWING: //... break; case LEADING: //... break; } } } finally { //... } }
public Vote lookForLeader() throws InterruptedException { //... try { HashMap recvset = new HashMap(); HashMap outofelection = new HashMap(); int notTimeout = finalizeWait; synchronized(this){ //给自己投票 logicalclock.incrementAndGet(); updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch()); } //将投票信息发送给集群中的每个服务器 sendNotifications(); //循环,如果是竞选状态一直到选举出结果 while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){ Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS); //没有收到投票信息 if(n == null){ if(manager.haveDelivered()){ sendNotifications(); } else { manager.connectAll(); } //... } //收到投票信息 else if (self.getCurrentAndNextConfigVoters().contains(n.sid)) { switch (n.state) { case LOOKING: // 判断投票是否过时,如果过时就清除之前已经接收到的信息 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; } } if (n == null) { self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING: learningState()); Vote endVote = new Vote(proposedLeader, proposedZxid, proposedEpoch); leaveInstance(endVote); return endVote; } } break; case OBSERVING: //忽略 break; case FOLLOWING: case LEADING: //如果是同一轮投票 if(n.electionEpoch == logicalclock.get()){ recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch)); //判断是否投票结束 if(termPredicate(recvset, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state)) && checkLeader(outofelection, n.leader, n.electionEpoch)) { self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING: learningState()); Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch); leaveInstance(endVote); return endVote; } } //记录投票已经完成 outofelection.put(n.sid, new Vote(n.leader, IGNOREVALUE, IGNOREVALUE, n.peerEpoch, n.state)); if (termPredicate(outofelection, new Vote(n.leader, IGNOREVALUE, IGNOREVALUE, n.peerEpoch, n.state)) && checkLeader(outofelection, n.leader, IGNOREVALUE)) { synchronized(this){ logicalclock.set(n.electionEpoch); self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING: learningState()); } Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch); leaveInstance(endVote); return endVote; } break; default: //忽略 break; } } else { LOG.warn("Ignoring notification from non-cluster member " + n.sid); } } return null; } finally { //... } }
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/69951287/viewspace-2664885/ ,如需转载,请注明出处,否则将追究法律责任。
注册时间:2019-10-19