网站优化要素,wordpress图片批量上传,做网站买一个域名就够了吗,更改网站标题zookeeper选举流程源码分析
选举的代码主要是在QuorumPeer.java这个类中。
它有一个内部枚举类#xff0c;用来表示当前节点的状态。 public enum ServerState {LOOKING, FOLLOWING, LEADING, OBSERVING;}LOOKING: 当前节点在选举过程中 FOLLOWING#xff1a;当前节点是从节…zookeeper选举流程源码分析
选举的代码主要是在QuorumPeer.java这个类中。
它有一个内部枚举类用来表示当前节点的状态。 public enum ServerState {LOOKING, FOLLOWING, LEADING, OBSERVING;}LOOKING: 当前节点在选举过程中 FOLLOWING当前节点是从节点 LEADING 当前节点是主节点 OBSERVING 当前节点是观察者状态这种状态的节点不参与选举的投票。 QuorumPeer有个run方法就是用来根据当前节点不同的状态进行不同的处理。
下面看下这段代码主要的框架 Overridepublic void run() {updateThreadName();LOG.debug(Starting quorum peer);// 这里是注册jmx消息不用关注//下面就是选举的框架代码了try {//running 表示当前节点的状态只要在运行过程中就会一直根据当前节点的状态进行不同的处理while (running) {//getPeerState()用来获取当前节点的状态就是上面提到的枚举类。//下面就会根据不同的状态进行不同的处理switch (getPeerState()) {case LOOKING:LOG.info(LOOKING);......//选举就是调用下面的这行代码来完成的。//后面我们也就单独就这个代码来进行分析setCurrentVote(makeLEStrategy().lookForLeader());......break;case OBSERVING:......//按照观察者的逻辑进行处理 break;case FOLLOWING:......//按照从节点的逻辑进行处理break;case LEADING:......//按照主节点的逻辑进行处理break;}start_fle Time.currentElapsedTime();}} finally {......}}上面代码的逻辑还是比较清楚的就是一直在这几种状态之间处理。
每种状态的处理逻辑基本都是如下 try {//处理业务逻辑正常情况下会一直在这里。//除非当前的状态逻辑已经处理完毕如LOOKING或者抛出了异常这时就需要重置状态} catch (Exception e) {LOG.warn(Unexpected exception,e);} finally {//重置状态updateServerState();}下面我们看看上面选举的这行代码 setCurrentVote(makeLEStrategy().lookForLeader());。这行代码会调用具体执行选举的类执行具体的选举操作并返回对应的投票信息并设置成当前的投票信息。
默认的选举的是FastLeaderElection对应的选举逻辑就在lookForLeader方法中。下面我们就直接去看看FastLeaderElection的lookForLeader方法吧。
选举的主要逻辑就是告诉其他节点。我是谁我选谁做为主节点。
public Vote lookForLeader() throws InterruptedException {......try {//recvset用来保存投票信息//key表示选民身份也就是这个票是谁投的(注意每个节点只会有一个有效的投票后面的投票会覆盖掉之前的投票)//value用来表示具体投票的内容HashMapLong, Vote recvset new HashMapLong, Vote();HashMapLong, Vote outofelection new HashMapLong, Vote();int notTimeout finalizeWait;synchronized(this){//每次投票前会先更新这个logicalclock逻辑时钟这个用来表示当前是第几次选举了对比投票信息的时候会用到这个很关键logicalclock.incrementAndGet();//首先给自己投一票updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());}LOG.info(New election. My id self.getId() , proposed zxid0x Long.toHexString(proposedZxid));//发送投票信息sendNotifications();
在上面的代码首先会把logicalclock1,表示当前是启动后的第几轮选取这个参数是保存在内存中的也就是每次启动都会从0开始。
那会不会出现节点之间logicalclock不同的情况呢这个情况是有可能会出现的。不过后面选举过程中相互发送消息也就会发送logicalclock,会和自己的logicalclock比较进行修正。
在开始选举的时候首先会给自己投一票。
会调用sendNotifications方法将投票者(自己)的信息和投票信息发出去。
会发送这些信息
proposedLeader : 选举的主节点proposedZxid: 选举的节点zxid,这个字段是long类型前面32 bit表示epoch,后面32bit表示事务idlogicalclock投票者的逻辑时钟QuorumPeer.ServerState.LOOKING投票者的状态(投票的状态肯定是looking)sid: 投票者的idproposedEpoch :选举节点的epoch,也就是proposedZxid的前面32 bit
下面看看具体的选举代码 //如果当前节点一直是looking且服务没有停止就会一直进行选举流程while ((self.getPeerState() ServerState.LOOKING) (!stop)){// 获取其他节点发送过来的消息Notification n recvqueue.poll(notTimeout,TimeUnit.MILLISECONDS);//如果没有收到消息就去检查下和其他节点的连接是否正常尽力使消息能发送。if(n null){......} // 验证收到消息的节点和它选举的主节点是否有效else if (validVoter(n.sid) validVoter(n.leader)) {//这里就会根据收到消息的节点状态进行分别进行处理// 比如自己是后加入进来的这时就已经有了leader节点对应的也就有follow节点// 也有可能大家都刚启动或者主节点挂掉了这时大家都会又会是looking状态switch (n.state) {// 如果对方节点是投票状态case LOOKING:// If notification current, replace and send messages out//首先比较logicalclock,如果对方的logicalclock比自己的大就修正自己的logicalclock同时清空自己的票箱重新计票if (n.electionEpoch logicalclock.get()) {logicalclock.set(n.electionEpoch);recvset.clear();// 这里会比较票的信息如果对方选的leader节点的比自己大就推举对方选的leader节点否则还是将票投给自己if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {updateProposal(n.leader, n.zxid, n.peerEpoch);} else {updateProposal(getInitId(),getInitLastLoggedZxid(),getPeerEpoch());}sendNotifications();// 如果自己的logicalclock 比对方的大直接忽略对方的票} else if (n.electionEpoch logicalclock.get()) {......break;// 如果logicalclock相等那就直接比较自己当前选出来的leader和对方选出来的leader进行比较如果自己的大就不做处理如果对方的大就更新自己的票重新投票} else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,proposedLeader, proposedZxid, proposedEpoch)) {updateProposal(n.leader, n.zxid, n.peerEpoch);sendNotifications();}//在这里将对方的票扔进投票箱// dont care about the version if its in LOOKING staterecvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));//这里就对投票进行统计了如果过半就要设置leader了不过在这之前会再等一个时间看看其他节点是否有选出更适合的leader。//如果没有那就设置对方节点选出来的leader为主节点对比下leader是不是自己如果是自己就将自己的状态修改为leader否则就修改成follow。同时保存当前leader信息if (termPredicate(recvset,new Vote(proposedLeader, proposedZxid,logicalclock.get(), proposedEpoch))) {// Verify if there is any change in the proposed leaderwhile((n recvqueue.poll(finalizeWait,TimeUnit.MILLISECONDS)) ! null){if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,proposedLeader, proposedZxid, proposedEpoch)){recvqueue.put(n);break;}}/** This predicate is true once we dont read any new* relevant message from the reception queue*/if (n null) {self.setPeerState((proposedLeader self.getId()) ?ServerState.LEADING: learningState());Vote endVote new Vote(proposedLeader,proposedZxid, logicalclock.get(), proposedEpoch);leaveInstance(endVote);return endVote;}}break;case OBSERVING://这种状态的节点是不参与投票的所以对它的发送的投票信息进行忽略。break;case FOLLOWING:case LEADING:
//如果对方是following或者leading,说明当前已经有主节点了在这里就直接统计票数信息并验证根据票数信息统计出来的leader节点和回应自己消息的自称leader节点 是不是同一个如果是同一个说明信息是吻合的就会去设置自己的节点状态。需要注意的是投票信息不但会发送给其他节点也会给自己发送。所以这里会判断对方节点是否是当前节点。......break;default:......}} else {......}}return null;} finally {......}}
比较节点大小也比较简单。
((newEpoch curEpoch) ||((newEpoch curEpoch) ((newZxid curZxid) || ((newZxid curZxid) (newId curId)))));首先比较epoch其次比较zxid最后比较myid。
myid就是我们在zookeeper每个节点中设置myid文件中对应的值。
zxid是两部分前32bit epoch后32 bit 事务序号。在一个节点成为leader节点后首先会将epoch的值1同时将事务序号设置成0。zxid是持久化写入文件的所以重启也不会丢失。
logicalclock在内存中所以每次启动都会从0开始。
给其他节点发送投票消息的时候也会给自己发送其他节点是通过网络发送给自己是直接放到接收投票信息的队列。
