首页  思维导图  详情

zookeeper

2023-02-17 21:59:30   4  举报





AI智能生成

zookeeper启动流程图

ZooKeeper流程图

作者其他创作

大纲/内容

loadDataBase();
zookeeper加载数据文件

zkDb.loadDataBase();

cnxnFactory.start();
启动NIO或者Netty

bootstrap.bind(localAddress);
绑定地址端口

收到数据
CnxnChannelHandler进行处理
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception

LeaderZooKeeperServer.setupRequestProcessors
初始化责任链

LeaderRequestProcessor

PrepRequestProcessor
构建数据，zxid

ProposalRequestProcessor
将消息分发给follower

CommitProcessor

SyncRequestProcessor
持久化消息

AckRequestProcessor

TobeAppliedRequestProcessor

FinalRequestProcessor

处理数据
cnxn.processMessage((ByteBuf) msg);

receiveMessage(buf);

zks.processPacket(this, bb);

submitRequest(si);

在LeaderRequestProcessor初始化
构建责任链处理
firstProcessor.processRequest(si);

放入队列
submittedRequests.add(request);

PrepRequestProcessor.start()

pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true);
构建zxid

request.zxid = zks.getZxid();
nextProcessor.processRequest(request);

ProposalRequestProcessor
sendPacket(pp);

遍历follower，发送数据 Leader.class
for (LearnerHandler f : forwardingFollowers) {
f.queuePacket(qp);
}

将消息放入队列

写数据文件，run方法
syncProcessor.processRequest(request);

flush(toFlush);

写文件
zks.getZKDatabase().commit();

AckRequestProcessor
nextProcessor.processRequest(i);

给自己发送一个ACk
leader.processAck(self.getId(), request.zxid, null);

加入集合
p.addAck(sid);

判断是否可以提交
boolean hasCommitted = tryToCommit(p, zxid, followerAddr);

qvAckset.getQuorumVerifier().containsQuorum(qvAckset.getAckset()

响应是否超过半数
return (ackSet.size() > half);

超过半数,发送给follower
commit(zxid);

构建commit类型数据包
QuorumPacket qp = new QuorumPacket(Leader.COMMIT, zxid, null, null);
sendPacket(qp);

inform(p);

发送数据包给Observer

自己commit
zk.commitProcessor.commit(p.request);

sendToNextProcessor(pending);

构建内存写数据
FinalRequestProcessor.processorRequest
rc = zks.processTxn(request);

createNode

回复监听事件
dataWatches.triggerWatch(path, Event.EventType.NodeCreated);
childWatches.triggerWatch(parentName.equals("") ? "/" : parentName,
Event.EventType.NodeChildrenChanged);

监听是一次性的
watchers = watchTable.remove(path);

case OpCode.getData

zks.getZKDatabase().getData(getDataRequest.getPath(), stat,
getDataRequest.getWatch() ? cnxn : null);

把watch放进去
dataWatches.addWatch(path, watcher);

处理完数据，回复客户端
cnxn.sendResponse(hdr, rsp, "response");

channel.writeAndFlush(Unpooled.wrappedBuffer(sendBuffer)).addListener(onSendBufferDoneListener);

startLeaderElection();
初始化选举相关数据

currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
生成当前节点

createElectionAlgorithm(electionType);

qcm = createCnxnManager();
初始化选举管理

listener.start();
启动监听

ss = new ServerSocket();
创建serverSocket
Socket client = ss.accept();
socket接收连接

handleConnection(sock, din);
处理连接信息

sid = din.readLong();
读取发送选票的机器id

如果机器id小于自己,就断开连接,并发送连接

大于就启动
SendWorker
RecvWorker

snedWorker像发送连接的sid发送选票

RecvWorker
将接收到选票放入recevQueue

le = new FastLeaderElection(this, qcm);
启动快速选举线程

启动WorkerSender
发送选票

process(m);
处理选票

如果是自己的sid，就放入自己recvQueue

addToSendQueue(bq, b);
将选票放入发送线程

connectOne(sid);
进行连接

启动WorkerReceiver
接收选票

self.getPeerState() == QuorumPeer.ServerState.LOOKING
如果当前自己还是Looking状态

是

已经选举出了leader

super.start();

case LOOKING:

logicalclock.incrementAndGet();
逻辑时钟+1

updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
初始化选票自己

sendNotifications();
将选票发送出去

Notification n = recvqueue.poll(notTimeout,
TimeUnit.MILLISECONDS);
从别人那里获取读取

如果为null创建连接

不为null

判断接收到的状态

也是looking的时候就进行选票PK

将胜出的选票再次发送

termPredicate(recvset,
new Vote(proposedLeader, proposedZxid,
logicalclock.get(), proposedEpoch))
判断是否超过半数

更新自己状态

case FOLLOWING:

follower.followLeader();

connectToLeader(leaderServer.addr, leaderServer.hostname);

建立Socket连接
sockConnect(sock, addr, Math.min(self.tickTime * self.syncLimit, remainingInitLimitTime));

syncWithLeader(newEpochZxid);

zk.startup();

创建责任链
setupRequestProcessors();

SyncRequestProcessor

SendAckRequestProcessor

FollowerRequestProcessor

死循环接收数据
QuorumPacket qp = new QuorumPacket();
while (this.isRunning()) {
readPacket(qp);
processPacket(qp);
}

case Leader.PROPOSAL:
fzk.logRequest(hdr, txn);

写入本机
syncProcessor.processRequest(request);

写回去
SendAckRequestProcessor.processRequest

case LEADING:

makeLeader(logFactory)
创建leader监听

leader.lead();

cnxAcceptor = new LearnerCnxAcceptor();
cnxAcceptor.start();

接收事件
s = ss.accept();

LearnerHandler.start();

startSendingPackets();

将消息发送出去
oa.writeRecord(p, "packet");

接收消息，并且进行反序列化
qp = new QuorumPacket();
ia.readRecord(qp, "packet");

case Leader.ACK:
leader.processAck(this.sid, qp.getZxid(), sock.getLocalSocketAddress());

startZkServer();

for (LearnerHandler f : getLearners()) {
                    // Synced set is used to check we have a supporting quorum, so only
                    // PARTICIPANT, not OBSERVER, learners should be used
                    if (f.synced() && f.getLearnerType() == LearnerType.PARTICIPANT) {
                        syncedSet.add(f.getSid());
                    }
                    f.ping();
                }
一直循环pingFollwer