首页  流程图  详情

5、zookeeper中leader选举过程

2021-01-28 14:58:09   0  举报





为你推荐

查看更多



Zookeeper leader选举原理

zookeeper源码之Leader选举

ZooKeeper-Leader选举源码（3.4.14）

zookeeper选举过程源码流程图

leader选举过程

Zookeeper Leader 选举

Zookeeper选举Leader多级队列

zk选举过程——Leader lead()方法

zookeeper选举Leader源码剖析

leader选举多层队列架构

仅限于zk的leader选举过程，其中包含两段选举

java

zookeeper

作者其他创作

大纲/内容

开始向其他节点同步数据

Messenger

消息发送队列，用于保存那些待发送的消息，按照SID进行分组

startLeaderElection();

初始线程组

myid

WorkerReceiver

发送心跳机制ping其他节点保持节点之间的活性

解析配置文件到内存中

sendqueue<ToSend>

接收线程，接收来自其他server的消息并处理这些消息，在选举过程没有停止的时候，接收线程就循环处理接收到的消息。处理消息的过程是：在外部消息队列recvQueue中取得一个消息，如果这个消息的发送端是个observer，那么立刻回复，将本server要选举的leader、当前状态、epoch信息发送过去。如果这个消息来自于一个participate，那么按本server的状态讨论，如果本server处于looking状态，那么将该消息放到本地接收队列中，同时判断发送方的状态，如果发送方也是looking状态并且其epoch值比较小，那么回复它一个消息告知本server的选举的leader信息。如果本server不是looking状态，而对方是looking状态的话，就回复一个消息告知本server所知道的leader信息。

super.start();

QuorumServer leaderServer = findLeader();

quorumPeer.setXXXX();

启动内嵌jetty服务

queueSendMap

④

purgeMgr.start();

args.length == 1 && config.isDistributed()

否

发送自己的选票结果

无论什么情况都将自己本地记录的leader修改成选举后的结果

sendqueue

启动Netty服务

启动等待新follower的线程

绑定业务处理CnxnServerChannelHandler

// 启动Leader线程ZooKeeperServer.setupRequestProcessors

选举监听

此处注意：getInitLastLoggedZxid()其实zxid是从ZKDatabase中的dataTree获取的，也就是未commit到dataTree中的zxid都是不算的

初始化选票（将自己作为初始化选票）

LearnerHandler.run()

同步Leader数据

SendWorker

接收的数据

设置自己的状态如果选举出来的id和自己是一样的话，那么就直接设置自己是leader

⑥

sendqueue.offer(notmsg);发送给应用层，交给其他的节点进行选举

有

config.parse(args[0]);

fle.start();

new SendWorker

初始化Netty

RecvWorker

s = ss.accept();

处理连接信息

加载本地数据

zkServer.sh

⑦

listener.start();

QuorumPeerMain.main服务器端启动类

1、先拿到recvQueue中的数据2、在循环调用senderWorkerMap将数据中的SendWorker发送给在zk上的其他节点

开始向其他节点进行同步数据

Listener.run();

返回选出的leader并设置到自己的属性中

接收消息队列作用：用于存储接收的所有选票

this.messenger = new Messenger(manager);

senderWorkerMap

switch (getPeerState())根据节点判断自身的状态

工作接收器作用：将用来收集其他zk节点的选票

zk_leader_1

2、sid == self.getId()

WorkerSender

makeLeader(logFactory)

作用：每一个与自身节点建立连接并【发送】数据的线程，从queueSendMap中获取需要发送的消息，且发送给对应的节点

没有

绑定端口

sendNotifications();

ServerBootstrap bootstrap = new ServerBootstrap()

发送的数据2

作用：主要用于记录zk服务于比自己小的节点<节点id，SendWorker>，zk只允许自己去连接比自己小的节点

接收选票器作用：每一个与自身节点建立连接并【接收】数据的线程。将数据汇总到recvQueue中去

new WorkerReceiver

FastLeaderElection

进行连接

ArrayBlockingQueue

createCnxnManager();

一般不会发生，发生的话也是进行警告，并不做什么事情

选票发送队列，用于保存待发送的选票。

startZkServer();

不进行业务处理，进入到下一次循环中去

开启WorkerSender与WorkerReceiver这两个线程，让他们开始发送与接收选票

主要是zk只允许myid大的去建立myid小的连接，不允许myid小的对myid大的进行创建连接

new RecvWorker

QuorumCnxManager

启动内嵌jetty服务，默认端口8080，用于监控整个zk状态

zk.loadData();

self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING: learningState());

connectOne(sid);

客户端命令

1、从recvQueue循环获得数据2、进行校验、解析数据、处理业务逻辑等3、判断是否是无权投票的机器3.1、是：就将数据放到发送数据队列，直接返回数据3.2、不是：放到recvqueue中，等待发送数据

发送消息队列

与Leader建立连接

recvqueue

接收的选举周期大于自己的选举周期适用于最后启动成功，再加入集群的机器

QuorumPeerMain.initializeAndRun

开始绑定地址进行与客户端连接ss = new ServerSocket();ss.bind(addr);setSockOpts(client);

sendqueue

while (this.isRunning()) { readPacket(qp); processPacket(qp);}

①

选举周期小于自己这种情况本身不操作

消息接收队列，用于存放那些从其他服务器接收到的消息。

1、创建SendWorker（消息发送者）、RecvWorker（消息接受者）2、给senderWorkerMap（发送工作者组）中放入sid（zk节点配置的myid）、SendWorker（消息发送者）3、给自己投一张leader选票4、启动SendWorker（消息发送者）、RecvWorker（消息接受者）线程

updateProposal（）

读取发送选票的ID

选举主逻辑LOOKING状态

②

makeFollower(logFactory)

将选票保存到发送消息队列sendqueue.offer(notmsg);

n.electionEpoch > logicalclock.get()

senderWorkerMap

quorumPeer.start();

totalOrderPredicate

if(n == null)

fh.start();

create /vico 666

queueSendMap

③

负责发送消息到其他server。发送消息的过程是：只要选举没有停止就从本地发送队列sendqueue中取一个消息并将它发送到指定的server。

重新去选举一次保证自己不会遗失选票如果不够半数的话，那就重新拉去，等到半数再进行选举

((newEpoch > curEpoch) || ((newEpoch == curEpoch) && ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))))其中选举为leader有3种情况：1、新的选举周期大于当前选举周期，选择选举周期大的2、选举周期相同时，选择zxid大的3、zxid相同时，选择id大的

创建选举算法

放入自己的选票箱中，用于后面进行计算过半选举

调用main方法进行启动

cnxAcceptor.start();

初始化zk配置

zk.startup();

makeLEStrategy().lookForLeader());

while(running)死循环，用于切换状态

sendNotifications();

recvqueue<Notification>

quorumPeer = getQuorumPeer()

ByteBuffer

logicalclock.incrementAndGet();

follower.followLeader();

CnxnChannelHandler.channelRead()

cnxAcceptor = new LearnerCnxAcceptor();

设置当前节点的状态

LearnerCnxAcceptor.run()

recvQueue

把选票发送给全部的节点

BinaryInputArchive.readRecord()

receiveConnection(client);

FastLeaderElection.lookForLeade()

startServerCnxnFactory();

n.electionEpoch < logicalclock.get()

逻辑类似

ServerCnxnFactory.createFactory()

QuorumCnxManager.run

ZooKeeperServerMain.main(args)

leader.lead();

选票接收队列，用于保存接收到的外部投票。

protocolVersion = din.readLong();

QuorumCnxManager

loadDataBase();

开启死循环用来接收Leader数据

作用：用于接收别的节点发送过来消息的队列

选票接收器。其会不断地从QuorumCnxManager中获取其他服务器发来的选举消息，并将其转换成一个选票，然后保存到recvqueue中，在选票接收过程中，如果发现该外部选票的选举轮次小于当前服务器的，那么忽略该外部投票，同时立即发送自己的内部投票。

pipeline.addLast(\"servercnxnfactory\

把自己注册到Leader中去

case LEADING

是

n.state判断选票的状态（是发送方的选票状态）

// 注册JMXManagedUtil.registerLog4jMBeans();

WorkerSender

NettyServerCnxnFactory

syncWithLeader(newEpochZxid);

初始化服务端连接对象，默认使用BIO，官方推荐使用Netty。如需Netty的话，在启动命令上添加参数-Dzookeeper.serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory

工作发送器作用：将这些选票按照sid发送给zk其他节点

1、读取发送选票的ID2、从senderWorkerMap中获取一次sid3、关闭这个连接4、自己重新建立与sid的连接

adminServer.start();

new WorkerSender

case FOLLOWING

接收follower数据开始线程进行处理

这种一般是已经选出leader了，然后机器再次启动后，自己投票给自己，其他机器收到后，会发已选出的leader给它，这个发送方的状态就是leader/follower

return endVote;

3、sid > self.getId()

获取本服务节点

1、sid < self.getId()

此处通过启动时是否有参数，判断是否为集群启动

1、将所有的选票过滤一遍，得到接受到的机器发送的最后的选票2、先按照配置的进行分组，再累计每个分组中的权重值3、统计出权重值最大的那个分组4、进行比对数值是否过半

重新统计一次看是否还有新的选票

case LOOKING

当前选票+1

设置属性注入zookeeper服务中去，其中包含较为重要的：1、dataLogDir：数据日志存储地址2、dataDir：数据存储地址3、electionAlg：默认选举方式3，快速选举4、myid：配置的myid5、ZKDatabase：初始化内存对象6、cnxnFactory：将上面配置好的服务对象，放到本节点中去

作用：用于自己给其他zk节点发送消息的队列<节点id，消息队列>，

获取选举的Leader Server

zk_following_2

选票发送器，不断地从sendqueue中获取待发送的选票，并将其传递到底层QuorumCnxManager中

开始进行选举

QuorumPeer.run();

this.electionAlg = createElectionAlgorithm(electionType);

如果本身是looking状态就一直循环

for (LearnerHandler f : getLearners()) { f.ping();}

ss = new ServerSocket();

bootstrap.bind(localAddress)

初始化选举数据管理器