zoukankan      html  css  js  c++  java
  • zookeeper原理解析-服务器端处理流程

    1)处理器链

        

    这部分内容我们主要讲解zookeeper请求在zookeeper server端的处理流程,对于不同角色的zookeeper具有不同的处理流程, ZookeepeerServer的start方法中会调用setupRequestProcessors()来初始化处理器链,它被子类覆写实现。

     

    1.      LeaderZooKeeperServer

    看如上代码主要建立了如下的两个处理器流链

    (1)    PrepRequestProcessor(线程) => ProposalRequestProcessor(调initialize) =>CommitProcessor(线程) => Leader.ToBeAppliedRequestProcessor=>FinalRequestProcessor

    (2)  ProposalRequestProcessor构造器设置另一处理器链, initialize方法启动SyncRequestProcessor线程 SyncRequestProcessor(线程)=> AckRequestProcessor

    2.      FollowerZooKeeperServer

    看如上代码主要建立了如下的两个处理器流链

    (1)  FollowerRequestProcessor  => CommitProcessor(线程)  =>FinalRequestProcessor(线程)

    (2)  SyncRequestProcessor(线程)=> SendAckRequestProcessor

     

    3.      ObserverZooKeeperServer

    (1)    ObserverRequestProcessor => CommitProcessor(线程)  =>FinalRequestProcessor(线程)

    (2)    SyncRequestProcessor(线程)=>  SendAckRequestProcessor

    (3)    看如上代码主要建立了如下的两个处理器流链

    2)处理器详解

     各个processor的主要功能

    1) PrepRequestProcessor

             如名字这个处理器主要功能是对请求进行预处理, 将client向server请求二进制数据反序列化成sever中请求操作。

    PrepRequestProcessor做为leader的第一个处理器,它的请求数据来源主要来自:

    (1)    Leader做一个zk服务接收客户端请求提交到PrepRequestProcessor的队列中

    (2)    作为集群的leader,在LearnerHanler.run方法中接收learner向leader发送的投票请求,消息类型为Leader.REQUEST

    PrepRequestProcessor的处理流程:

    (1)    对于非事物性请求:sync,exists, getData, getChildRen,ping, setWatches 这里只做session的检查看看是否超时

    (2)    对于事务请求:create, delete,setData,setAcl,check,multi,根据请求的类型创建不同的操作如:type=create è CreateRequest, type=delete èDeleteRequest  等等

    (3)    Zookeeper创建新的事务号zxid

    (4)    创建并设置事务请求的消息头

    (5)    反序列化网络client请求对象提取数据到事务对象的消息体中

    PrepRequestProcessor线程 { 

        run {

           1. submittedRequests.take()取出nio读取的请求

           2. 根据请求type构建对应的record对象,并将request中的ByteBuffer数据,反序列化到record中

           3. zks.getNextZxid() 生成一个新的事务号递增,作为zxid

           4. 根据cxid,zxid, sessionId等构建事务头TxnHeader

           5.1 对create/delete/setData/setACL/createSession/closeSession/check

                   1)checkSession,检测是否过期

                   2)创建对应的事务体CreateTxn/DeleteTxn等等

                     3)构建ChangeRecord加入到zks.outstandingChanges队列中去(FinalRequestProcessor会去处理,最终去改变数据)

           5.2 exists/getData 只是 checkSession

           6. 调下一个Processor

        }

     

    processRequest() {

           //单机版被zookeeper调,将nio读取的请求加入到submittedRequests中

        }

    }

    2)ProposalRequestProcessor

             ProposalRequestProcessor的处理逻辑相对比较简单

    (1)    转发到后面的处理器

    (2)    如果是事务性请求(请求头存在的话),leader向follower发起操作请求,超过一半才算成功响应客户端

    (3)    如果是事务性请求,调用leader的同步处理器流程

    3)CommitProcessor

       这个处理器逻辑还是有点小复杂的, leader和learner都需要用到这个处理器

    3.1) 对于非事务性的操作(查询,exist等)直接回把请求转到下一个处理器处理

    3.2) leader 对于事务性操作(create, setData等)请求,CommitProcessor线程任务会hold在这里,leader中ProposalRequestProcessor处理器会将请求提案发送给所有的followers, followers响应leader,然后leader中LearnerHandler会调processAck处理响应,当超过半数的时候将调CommitProcessor.commit()方法提交请求, 紧接着CommitProcessor将请求传递到下一个处理器处理

    3.2) learner对于事务性操作(create, setData等)请求CommitProcessor线程任务会hold在这里, FollowerRequestProcessor或者ObserverRequestProcessor调CommitProcessor将请求提交队列之后会立刻向leader发送事务操作提案,Follower接收到leader的commit消息或者Observer接收到leader的inform消息它们会向CommitProcessor提交请求,紧接着CommitProcessor将请求传递到下一个处理器处理

    伪代码:

    CommitProcessor{

       run() {

           1. toProcess需要交予下一个Processor的,先都交给下一个

           2. nextPending请求时对于事务操作的,有一个不为空一直循环直到有commit过来

           3. queuedRequests.size() == 0&& committedRequests.size() > 0 follower observer接收commit ,加入到toProcess集合中去

           4. nextPending != null&&  committedRequests.size() > 0  leader发起投票请求,并接收follower反馈的, 加入到toProcess集合中去

           5. nextPending == null 前面循环

           6.如果是请求reqeust是事务操作赋给nextPending对象

           7.如果不是加入到toProcess集合中去

            

           //这里主要通过nextPending对象控制请求响应的顺序

        }

        commit(Request){

           将request添加到committedRequests队列中去

        }

        processRequest(Request) {

           由上游处理器调用,将request对象添加到queuedRequests请求队列中

       }

    }

    4)ToBeAppliedRequestProcessor

             这个处理器的逻辑比较简单

    1)   将请求转发给一下个处理器,必须是FinalRequestProcessor

    2)   其实leader在走到这个处理器之前会在CommitProcessor中hod一会等到follower反馈在到这,follower反馈后leader的LearnerHandler的processAck会将请求加入toBeApplied集合,所以在这里对于事务请求一定会在toBeApplied中有对应的移除调,如果没有ConcurrentLinkedQueue直接会抛NoSuchElementException异常

    5)FinalRequestProcessor

       这个处理器是最后一个处理器,真正去执行事务操作更改dataree的数据。

    1)   调底层修改数据zks.processTxn(hdr, txn)

    2)   将请求加入到committedLog集合中

    3)   构建请求的响应,响应客户端

    伪代码:

    FinalRequestProcessor{

          processRequest(Request request) {

               //zks.outstandingChanges这个玩意起什么作用,一直没弄清楚

               1.事务头不为空,是事务类操作 {

                    zks.processTxn(hdr,txn) //zkServer处理事务操作

                }

               如果是closeSesion,无需生成响应

               根据请求类型(request.type)生成响应,并调NioServerCnxn.sendResponse写入chanel通道

          }

    }

    6) SyncRequestProcessor

    这个处理器用来将请求记录到txLog文件中,通过批量刷盘的方式来提升io的性能,这里请求只有被写入到本地磁盘后,才会被传递到下一个处理器

    下面看一下伪代码:

    SyncRequestProcessor线程 {

        run() {

            //flush的时间点:1. queuedRequests为空 2.toFlush.size() > 1000

            //生成新的snapshot

            调zks.getZKDatabase().append(si)添加一条事务日志

               1)成功:是事务类操作有事务头, 根据规则判断是否需要生一个新的snapshot,加入到toFlush的集合中    

               2)失败:没有事务头TxnHeader, 优化直接调下一个Processor

        }

        flush() {

            zks.getZKDatabase().commit();同步到本snapshot

            然后循环调下一个Processor

        }

        processRequest() {

           加入到阻塞队列queuedRequests中, 让同步线程自己处理

        }

    }

    7) AckRequestProcessor 

    被ProposalRequestProcessor调用, leader自己做一次投票的成功响应

            

    8) SendAckRequestProcessor

             对于leader投票请求的发送响应

    3)交互图

      

    1. 下面就用一张图来说明Leader端的处理器链的交互过程

    2. 下面就用一张图来说明Follower(Observer类似)端的处理器链的交互过程

  • 相关阅读:
    session判断重复提交
    logback日志配置
    quartz动态job工具类 serviceh注入问题
    mysql10061登录失败错误解决方案
    安装步骤
    无法启动此程序,因为计算机丢失MSVCP120.dll
    Bigdecimal: Non-terminating decimal expansion; no exact representable decimal result.
    replace()函数用法
    随机读取表中一条数据
    oracle table()函数
  • 原文地址:https://www.cnblogs.com/wxd0108/p/6251740.html
Copyright © 2011-2022 走看看