zookeeper

zookeeperquorumPeer.start();

zkDb.loadDataBase();

loadDataBase();zookeeper加载数据文件

FinalRequestProcessor

TobeAppliedRequestProcessor

AckRequestProcessor

SyncRequestProcessor持久化消息

CommitProcessor

ProposalRequestProcessor将消息分发给follower

PrepRequestProcessor构建数据，zxid

LeaderRequestProcessor

LeaderZooKeeperServer.setupRequestProcessors初始化责任链

放入队列submittedRequests.add(request);

PrepRequestProcessor.start()

将消息放入队列

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

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

加入集合p.addAck(sid);

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

发送数据包给Observer

inform(p);

sendToNextProcessor(pending);

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

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

createNode

case OpCode.getData

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

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

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

AckRequestProcessornextProcessor.processRequest(i);

flush(toFlush);

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

ProposalRequestProcessorsendPacket(pp);

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

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

submitRequest(si);

receiveMessage(buf);

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

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

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

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

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

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

RecvWorker将接收到选票放入recevQueue

大于就启动SendWorkerRecvWorker

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

listener.start();启动监听

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

connectOne(sid);进行连接

process(m);处理选票

启动WorkerSender发送选票

是

已经选举出了leader

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

启动WorkerReceiver接收选票

createElectionAlgorithm(electionType);

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

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

sendNotifications();将选票发送出去

如果为null创建连接

将胜出的选票再次发送

更新自己状态

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

判断接收到的状态

不为null

case LOOKING:

创建责任链setupRequestProcessors();

FollowerRequestProcessor

SendAckRequestProcessor

SyncRequestProcessor

zk.startup();

syncWithLeader(newEpochZxid);

写回去SendAckRequestProcessor.processRequest

写入本机syncProcessor.processRequest(request);

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

follower.followLeader();

case FOLLOWING:

makeLeader(logFactory)创建leader监听

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

startSendingPackets();

LearnerHandler.start();

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

startZkServer();

leader.lead();

case LEADING:

super.start();