zookeeper源码分析二FASTLEADER选举算法

如何在zookeeper集群中选举出一个leader,zookeeper使用了三种算法,具体使用哪种算法,在配置文件中是可以配置的,对应的配置项是”electionAlg”,其中1对应的是LeaderElection算法,2对应的是AuthFastLeaderElection算法,3对应的是FastLeaderElection算法.默认使用FastLeaderElection算法.其他两种算法我没有研究过,就不多说了.

要理解这个算法,最好需要一些paxos算法的理论基础.

1) 数据恢复阶段
首先,每个在zookeeper服务器先读取当前保存在磁盘的数据,zookeeper中的每份数据,都有一个对应的id值,这个值是依次递增的,换言之,越新的数据,对应的ID值就越大.

2) 首次发送自己的投票值
在读取数据完毕之后,每个zookeeper服务器发送自己选举的leader,这个协议中包含了以下几部分的数据:
1)所选举leader的id(就是配置文件中写好的每个服务器的id) ,在初始阶段,每台服务器的这个值都是自己服务器的id,也就是它们都选举自己为leader.
2) 服务器最大数据的id,这个值大的服务器,说明存放了更新的数据.
3)逻辑时钟的值,这个值从0开始递增,每次选举对应一个值,也就是说:如果在同一次选举中,那么这个值应该是一致的 2)逻辑时钟值越大,说明这一次选举leader的进程更新.
4) 本机在当前选举过程中的状态,有以下几种:LOOKING,FOLLOWING,OBSERVING,LEADING,顾名思义不必解释了吧.

每台服务器将自己服务器的以上数据发送到集群中的其他服务器之后,同样的也需要接收来自其他服务器的数据,它将做以下的处理:
1) 如果所接收数据服务器的状态还是在选举阶段(LOOKING 状态),那么首先判断逻辑时钟值,又分为以下三种情况:
a) 如果发送过来的逻辑时钟大于目前的逻辑时钟,那么说明这是更新的一次选举,此时需要更新一下本机的逻辑时钟值,同时将之前收集到的来自其他服务器的选举清空,因为这些数据已经不再有效了.然后判断是否需要更新当前自己的选举情况.在这里是根据选举leader id,保存的最大数据id来进行判断的,这两种数据之间对这个选举结果的影响的权重关系是:首先看数据id,数据id大者胜出;其次再判断leader id,leader id大者胜出.然后再将自身最新的选举结果(也就是上面提到的三种数据广播给其他服务器).代码如下:

if (n.epoch > logicalclock) {
logicalclock = n.epoch;
recvset.clear();
if(totalOrderPredicate(n.leader, n.zxid,
getInitId(), getInitLastLoggedZxid()))
updateProposal(n.leader, n.zxid);
else
updateProposal(getInitId(),
getInitLastLoggedZxid());
sendNotifications();

其中的totalOrderPredicate函数就是根据发送过来的封包中的leader id,数据id来与本机保存的相应数据进行判断的函数,返回true说明需要更新数据,于是调用updateProposal函数更新数据

b) 发送过来数据的逻辑时钟小于本机的逻辑时钟
说明对方在一个相对较早的选举进程中,这里只需要将本机的数据发送过去就是了

c) 两边的逻辑时钟相同,此时也只是调用totalOrderPredicate函数判断是否需要更新本机的数据,如果更新了再将自己最新的选举结果广播出去就是了.

三种情况的处理完毕之后,再处理两种情况:
1)服务器判断是不是已经收集到了所有服务器的选举状态,如果是那么根据选举结果设置自己的角色(FOLLOWING还是LEADER),然后退出选举过程就是了.
2)即使没有收集到所有服务器的选举状态,也可以判断一下根据以上过程之后最新的选举leader是不是得到了超过半数以上服务器的支持,如果是,那么尝试在200ms内接收一下数据,如果没有新的数据到来,说明大家都已经默认了这个结果,同样也设置角色退出选举过程.
代码如下:

/*
* Only proceed if the vote comes from a replica in the
* voting view.
*/
if(self.getVotingView().containsKey(n.sid)){
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.epoch));

//If have received from all nodes, then terminate
if ((self.getVotingView().size() == recvset.size()) &&
(self.getQuorumVerifier().getWeight(proposedLeader) != 0)){
self.setPeerState((proposedLeader == self.getId()) ?
ServerState.LEADING: learningState());
leaveInstance();
return new Vote(proposedLeader, proposedZxid);

} else if (termPredicate(recvset,
new Vote(proposedLeader, proposedZxid,
logicalclock))) {

// Verify if there is any change in the proposed leader
while((n = recvqueue.poll(finalizeWait,
TimeUnit.MILLISECONDS)) != null){
if(totalOrderPredicate(n.leader, n.zxid,
proposedLeader, proposedZxid)){
recvqueue.put(n);
break;
}
}

/*
* This predicate is true once we don‘t read any new
* relevant message from the reception queue
*/
if (n == null) {
self.setPeerState((proposedLeader == self.getId()) ?
ServerState.LEADING: learningState());
if(LOG.isDebugEnabled()){
LOG.debug("About to leave FLE instance: Leader= "
+ proposedLeader + ", Zxid = " +
proposedZxid + ", My id = " + self.getId()
+ ", My state = " + self.getPeerState());
}

leaveInstance();
return new Vote(proposedLeader,
proposedZxid);
}
}
}

2) 如果所接收服务器不在选举状态,也就是在FOLLOWING或者LEADING状态
做以下两个判断:
a) 如果逻辑时钟相同,将该数据保存到recvset,如果所接收服务器宣称自己是leader,那么将判断是不是有半数以上的服务器选举它,如果是则设置选举状态退出选举过程
b) 否则这是一条与当前逻辑时钟不符合的消息,那么说明在另一个选举过程中已经有了选举结果,于是将该选举结果加入到outofelection集合中,再根据outofelection来判断是否可以结束选举,如果可以也是保存逻辑时钟,设置选举状态,退出选举过程.
代码如下:

if(n.epoch == logicalclock){
recvset.put(n.sid, new Vote(n.leader, n.zxid, n.epoch));
if((n.state == ServerState.LEADING) ||
(termPredicate(recvset, new Vote(n.leader,
n.zxid, n.epoch, n.state))
&& checkLeader(outofelection, n.leader, n.epoch)) ){
self.setPeerState((n.leader == self.getId()) ?
ServerState.LEADING: learningState());

leaveInstance();
return new Vote(n.leader, n.zxid);
}
}

outofelection.put(n.sid, new Vote(n.leader, n.zxid,
n.epoch, n.state));

if (termPredicate(outofelection, new Vote(n.leader,
n.zxid, n.epoch, n.state))
&& checkLeader(outofelection, n.leader, n.epoch)) {
synchronized(this){
logicalclock = n.epoch;
self.setPeerState((n.leader == self.getId()) ?
ServerState.LEADING: learningState());
}
leaveInstance();
return new Vote(n.leader, n.zxid);
}

break;
}
}

以一个简单的例子来说明整个选举的过程.
假设有五台服务器组成的zookeeper集群,它们的id从1-5,同时它们都是最新启动的,也就是没有历史数据,在存放数据量这一点上,都是一样的.假设这些服务器依序启动,来看看会发生什么.
1) 服务器1启动,此时只有它一台服务器启动了,它发出去的报没有任何响应,所以它的选举状态一直是LOOKING状态
2) 服务器2启动,它与最开始启动的服务器1进行通信,互相交换自己的选举结果,由于两者都没有历史数据,所以id值较大的服务器2胜出,但是由于没有达到超过半数以上的服务器都同意选举它(这个例子中的半数以上是3),所以服务器1,2还是继续保持LOOKING状态.
3) 服务器3启动,根据前面的理论分析,服务器3成为服务器1,2,3中的老大,而与上面不同的是,此时有三台服务器选举了它,所以它成为了这次选举的leader.
4) 服务器4启动,根据前面的分析,理论上服务器4应该是服务器1,2,3,4中最大的,但是由于前面已经有半数以上的服务器选举了服务器3,所以它只能接收当小弟的命了.
5) 服务器5启动,同4一样,当小弟.

以上就是fastleader算法的简要分析,还有一些异常情况的处理,比如某台服务器宕机之后的处理,当leader宕机之后的处理等等,后面再谈

时间: 2024-08-10 14:51:19

zookeeper源码分析二FASTLEADER选举算法的相关文章

Zookeeper 源码分析-启动

Zookeeper 源码分析-启动 博客分类: Zookeeper 本文主要介绍了zookeeper启动的过程 运行zkServer.sh start命令可以启动zookeeper.入口的main函数在类中QuorumPeerMain. main函数主要调用了runFromConfig函数,创建了QuorumPeer对象,并且调用了start函数,从而启动了zookeeper. Java代码   public class QuorumPeerMain { protected QuorumPeer

netty 源码分析二

以服务端启动,接收客户端连接整个过程为例分析, 简略分为 五个过程: 1.NioServerSocketChannel 管道生成, 2.NioServerSocketChannel 管道完成初始化, 3.NioServerSocketChannel注册至Selector选择器, 4.NioServerSocketChannel管道绑定到指定端口,启动服务 5.NioServerSocketChannel接受客户端的连接,进行相应IO操作 Ps:netty内部过程远比这复杂,简略记录下方便以后回忆

[Android]Volley源码分析(二)Cache

Cache作为Volley最为核心的一部分,Volley花了重彩来实现它.本章我们顺着Volley的源码思路往下,来看下Volley对Cache的处理逻辑. 我们回想一下昨天的简单代码,我们的入口是从构造一个Request队列开始的,而我们并不直接调用new来构造,而是将控制权反转给Volley这个静态工厂来构造. com.android.volley.toolbox.Volley: public static RequestQueue newRequestQueue(Context conte

哇!板球 源码分析二

游戏主页面布局 创建屏下Score标签 pLabel = CCLabelTTF::create("Score", "Arial", TITLE_FONT_SIZE); //分数标签 //设置标签字体的颜色 pLabel->setColor (ccc3(0, 0, 0)); //设置文本标签的位置 pLabel->setPosition ( ccp ( SCORE_X, //X坐标 SCORE_Y //Y坐标 ) ); //将文本标签添加到布景中 this

OpenStack_Swift源码分析——Ring的rebalance算法源代码详细分析

今天有同学去百度,带回一道面试题,和大家分享一下: 打印: n=1 1 n=2 3 3 2 4 1 1 4 5 5 n=3 7 7 7 7 6 8 3 3 2 6 8 4 1 1 6 8 4 5 5 5 8 9 9 9 9 提供一段参考程序: <pre name="code" class="cpp">// ConsoleApplication1.cpp: 主项目文件. #include "stdafx.h" #include &quo

zookeeper源码分析之五服务端(集群leader)处理请求流程

leader的实现类为LeaderZooKeeperServer,它间接继承自标准ZookeeperServer.它规定了请求到达leader时需要经历的路径: PrepRequestProcessor -> ProposalRequestProcessor ->CommitProcessor -> Leader.ToBeAppliedRequestProcessor ->FinalRequestProcessor 具体情况可以参看代码: @Override protected v

baksmali和smali源码分析(二)

这一节,主要介绍一下 baksmali代码的框架. 我们经常在反编译android apk包的时候使用apktool这个工具,其实本身这个工具里面对于dex文件解析和重新生成就是使用的baksmali 和smali这两个jar包其中 baksmali是将 dex文件转换成便于阅读的smali文件的,具体使用命令如下:java -jar baksmali.jar classes.dex -o myout其中myout是输出的文件夹 而smali是将smali文件重新生成回 dex文件的具体使用的命

zookeeper源码分析之一服务端处理请求流程

上文: zookeeper源码分析之一服务端启动过程 中,我们介绍了zookeeper服务器的启动过程,其中单机是ZookeeperServer启动,集群使用QuorumPeer启动,那么这次我们分析各自一下消息处理过程: 前文可以看到在 1.在单机情况下NettyServerCnxnFactory中启动ZookeeperServer来处理消息: public synchronized void startup() { if (sessionTracker == null) { createSe

【梦幻连连连】源码分析(二)

转载请注明出处:http://blog.csdn.net/oyangyufu/article/details/24736711 GameLayer场景界面效果: 源码分析: GameLayer场景初始化,主要是初始化加载界面及背景音乐 bool GameLayer::init() { float dt=0.0f; if ( !CCLayerColor::initWithColor(ccc4(255, 255, 255, 255))) { return false; } this->initLoa