Lengxf_zookpeer服务端与客户端源码流程

Main loop主逻辑阻塞式自旋

sendNotifications()发送选票初次推荐自己为leader将消息存放至ssendqueue.offer(notmsg)

OBSERVING

true

判断当前节点是否为本集群!getView().containsKey(myid)

不做任何操作

setCurrentVote(makeLEStrategy().lookForLeader())设置快速选举的选票

ServerState.LEADING

真实的投票逻辑处理选票

sendNotifications()选票信息交换sendqueue.offer(notmsg)

getInitId()初始选票默认为自身

ZooKeeperServerMain.main(args)单机模式启动

wrThread.start()启动WorkerReceiver

本节点数据直接存放至本服务的recvQueue中

有数据

1.接收连接receiveConnection2.处理连接handleConnection

wsThread.start()启动WorkerSender

如果得到的消息为无效选票则直接将本节点的状态返回给发送方

fasle

updateProposal修改选票

sendqueue.offer(notmsg)把消息投递到发送队列中

LEADING

服务默认启动状态为LOOKING会把自己作为初次选票投递对象

异常中断程序

LOOKING模式主逻辑

单机/集群config.isDistributed()

通常选用netty

reconfigFlagClear()重新配置的额标志位

与其他节点建立链接

ServerState.LOOKING节点启动时默认为此状态

LOOKING

启动线程主逻辑super.start()

readonlymode.enabled只读模式逻辑通常情况不会走此分支逻辑

ServerState.OBSERVING

false

FOLLOWING

启动JVM暂停监控startJvmPauseMonitor()

getPeerEpoch()选举周期

选举周期对齐后与其他节点的PK选票逻辑return ((newEpoch > curEpoch)                || ((newEpoch == curEpoch)                && ((newZxid > curZxid)                || ((newZxid == curZxid)                && (newId > curId)))));

MetricsProviderMetricsProviderBootstrap.startMetricsProvider探针识别程序是否可正常使用然后暴露服务

voteSet.hasAllQuorums()得票超过半数则节点变为leader

sid与myid比较只能id向小进行建立链接若id相等则说明配置错误若sid<myid则关闭链接然后反向链接

启动管理员服务adminServer.start()

SendWorker

DatadirCleanupManager#start()支线逻辑异步启动数据清理管理器

进入自旋逻辑直到选票成功while ((self.getPeerState() == ServerState.LOOKING) && (!stop))

队列无数据

!validVoter(response.sid)判断是否为有效选票

self.getPeerState() == QuorumPeer.ServerState.LOOKING判断自身是否为LOOKING状态

(ackstate == QuorumPeer.ServerState.LOOKING)     && (n.electionEpoch < logicalclock.get())其他节点为LOOKING状态且选举周期落后则把本节点信息返回

设置本节点的leader为接收到的消息

BlockingQueue<ByteBuffer> bq = queueSendMap.get(sid)消费此队列消息然后通过send()发送给其他节点

启动服务工厂startServerCnxnFactory()

启动leader选举startLeaderElection()

ServerState.FOLLOWING

this.messenger.start()启动选举

getInitLastLoggedZxid()最大的日志id

其他节点数据存放至queueSendMap通过SendWorker发送至其他节点

接收到其他follower发来的消息时 直接把本节点的leader设置为接收到的消息中所携带的leader信息

若选票节点为LOOKING状态会先判断是否需要重新选举leader然后把本节点的选票信息返回可以理解为把本节点的leader信息返回

ManagedUtil.registerLog4jMBeans()注册日志管理器

RecvWorker

加载基础数据loadDataBase()

recvqueue.offer(n)把冲票重新投递到recvqueue中

远程通信

QuorumPeerMain服务端启动类

异步接收来自其他节点连接QuorumCnxManager.ListenerListenerHandler#acceptConnections()

集群模式启动

默认使用NIO工厂

quorumPeer.start()启动当前节点quorumPeer.join()阻塞主线程