入口#
书接上篇博客中的ZK集群启动后完成数据的统一性恢复后,来到启动ZkServer的逻辑,接下来的重点工作就是启动不同角色的对应的不同的处理器Processor
如上图查看ZooKeeperServer的继承图,三种不同的角色有不同的ZooKeeperServer的实现逻辑类
三者启动时,都将会来到ZooKeeperServer.java中的startUp()方法中,源码如下,但是,不同的角色针对setupRequestProcessors();进行了不同的重写,所以本篇博客的重点即使看一下他们是如何重写的
public synchronized void startup() { if (sessionTracker == null) { // todo 创建session计时器 createSessionTracker(); } // todo 开启计时器 startSessionTracker(); // todo 设置请求处理器， zookeeper中存在不同的请求处理器, 就在下面 setupRequestProcessors(); //todo 是一个为应用程序、设备、系统等植入管理功能的框架。 //todo JMX可以跨越一系列异构操作系统平台、系统体系结构和网络传输协议，灵活的开发无缝集成的系统、网络和服务管理应用 registerJMX(); // todo 修改状态 --> running setState(State.RUNNING); // todo 唤醒所有线程, 因为前面有一个线程等待处理器 睡了一秒 notifyAll(); }
Leader重写setupRequestProcessors#
源码如下: 可以看到它初始化的处理器的个数
- PrepRequestProcessor (checkAcl 构造tnx)
- ProposalRequestProcessor (发起提议)
- CommitProcessor (提交提议)
- ToBeAppliedRequestProcessor
- FinalRequestProcessor (响应客户端,更新内存)
SyncRequestProcessor(单独开启的,他是一个线程) 作用: 持久化txn
RequestProcessor finalProcessor = new FinalRequestProcessor(this); RequestProcessor toBeAppliedProcessor = new Leader.ToBeAppliedRequestProcessor( finalProcessor, getLeader().toBeApplied); commitProcessor = new CommitProcessor(toBeAppliedProcessor, Long.toString(getServerId()), false, getZooKeeperServerListener()); commitProcessor.start(); ProposalRequestProcessor proposalProcessor = new ProposalRequestProcessor(this, commitProcessor); proposalProcessor.initialize(); firstProcessor = new PrepRequestProcessor(this, proposalProcessor); ((PrepRequestProcessor)firstProcessor).start();
Follower重写setupRequestProcessors#
- FollowerRequestProcessor
- CommitProcessor
- SendAckRequestProcessor
- FinalRequestProcessor
SyncRequestProcessor(单独开启的,他是一个线程) 作用: 持久化txn
RequestProcessor finalProcessor = new FinalRequestProcessor(this); commitProcessor = new CommitProcessor(finalProcessor, Long.toString(getServerId()), true, getZooKeeperServerListener()); commitProcessor.start(); firstProcessor = new FollowerRequestProcessor(this, commitProcessor); ((FollowerRequestProcessor) firstProcessor).start(); syncProcessor = new SyncRequestProcessor(this, new SendAckRequestProcessor((Learner) getFollower())); syncProcessor.start();
Observer重写setupRequestProcessors#
- ObserverRequestProcessor
- CommitProcessor
- FinalRequestProcessor
通过配置判断是否添加
SyncRequestProcessor来持久化它的事务
RequestProcessor finalProcessor = new FinalRequestProcessor(this); commitProcessor = new CommitProcessor(finalProcessor, Long.toString(getServerId()), true, getZooKeeperServerListener()); commitProcessor.start(); firstProcessor = new ObserverRequestProcessor(this, commitProcessor); ((ObserverRequestProcessor) firstProcessor).start(); // todo 通过这个判断控制需不需要Observer 对事务进行持久化 if (syncRequestProcessorEnabled) { syncProcessor = new SyncRequestProcessor(this, null); syncProcessor.start(); }
实验1: Leader接受到写请求#
直接给出当Leader接收到请求时,request在集群中各个处理器中的运行流程图
通过上图看,当leader接收到请求后,request肯定会依次流经它的处理器,PrepRequestProcessor-->ProposalRequestProcessor
在ProposalRequestProcessor处理器中,同样是直接将request提交给CommitProcessor,但是同样会被阻塞住
接着在request被Leader通过原子广播,告诉所有的Follower这个request
原子广播之后自己会立即使用SyncRequestProcessor完成持久化
同时Follower接受到request之后,也会使用他们自己的SyncRequestProcess进行持久化,完成持久化后就给Leader的LearnerHandler发送ACK确认消息,在这个LearnerHandler会存在过半检查机制,没当一个Follower发送一个ACK,就触发检查一次,直到达到一半以上,就会触发notify(),然后leader刚刚在commitProcessor中wait(),等待提交的函数就会被唤醒,由leader广播commit,全体learner进行commit,达成数据的一致性
实验2: Follower或Observer接收到写请求#
直接给出当Follower或者Observer接收到请求时,request在集群中各个处理器中的运行流程图
通过上面图可以看到,当Follower或者Observer接收到请求后会首先会提交给本地的commitProcessor处理器,但是不会立刻commit事务,而是将request转发给Leader的第一个处理器,再之后就和上面的图同样的处理流程
