zookeeper源码分析-Processor保证数据一致性

Posted 2023-02-06 suixinm

 

zookeeper源码分析，Processor保证数据一致性

入口

书接上篇博客中的ZK集群启动后完成数据的统一性恢复后,来到启动ZkServer的逻辑,接下来的重点工作就是启动不同角色的对应的不同的处理器Processor

如上图查看ZooKeeperServer的继承图,三种不同的角色有不同的ZooKeeperServer的实现逻辑类

三者启动时,都将会来到ZooKeeperServer.java中的startUp()方法中,源码如下,但是,不同的角色针对setupRequestProcessors();进行了不同的重写,所以本篇博客的重点即使看一下他们是如何重写的

Copy

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#

源码如下: 可以看到它初始化的处理器的个数

1. PrepRequestProcessor (checkAcl 构造tnx)
2. ProposalRequestProcessor (发起提议)
3. CommitProcessor (提交提议)
4. ToBeAppliedRequestProcessor
5. FinalRequestProcessor (响应客户端,更新内存)

SyncRequestProcessor(单独开启的,他是一个线程) 作用: 持久化txn

Copy

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#

1. FollowerRequestProcessor
2. CommitProcessor
3. SendAckRequestProcessor
4. FinalRequestProcessor

SyncRequestProcessor(单独开启的,他是一个线程) 作用: 持久化txn

Copy

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#

1. ObserverRequestProcessor
2. CommitProcessor
3. FinalRequestProcessor

通过配置判断是否添加SyncRequestProcessor来持久化它的事务

Copy

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的第一个处理器,再之后就和上面的图同样的处理流程