ZK集群中通过Processor保证数据一致性源码阅读

入口

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

原文地址：https://www.cnblogs.com/ZhuChangwu/p/11619978.html