3. Dubbo服务提供端请求处理流程

Posted 2023-04-26

参考技术A 从时序图上不难看出,服务提供端对请求的处理先通过处理器责任链一层一层处理,然后找到需要调用的服务实现类的代理Invoker进行调用，再将响应发送到调用方。

调用NettyServer#handler处理请求

Dubbo默认的底层网络通信使用的是Netty,NettyServer使用两级线程池,EventLoopGroup(boss),主要用来接收客户端连接请求,并把完成TCP三次握手的连接分发给EventLoopGroup(worker)处理,根据请求的消息是被I/O线程处理还是被业务线程池处理,Dubbo提供了几种线程模型,AllDispatcher是默认线程模型,AllChannelHandler是对应的处理器.

AllDispatcher:将所有消息(请求,响应,心跳,连接事件,断开事件)都派发到业务线程池.

ChannelEventRunnable处理消息的线程

对消息进行解码,解码后交由后续处理器处理

对消息进行分发处理

处理消息,并返回响应

从inv中获取 path 、version、group,从channel中获取调用的端口,组装serviceKey ,找到缓存在exporterMap的DubboExporter,进而获取服务实现类代理对象Invoker

调用代理类,处理请求,根据返回类型组装相应的Result,如果是开启了异步上下文或者返回值为CompletableFuture类型,则返回AsyncRpcResult,
否则返回同步返回结果RpcResult

至此,Dubbo提供端请求处理流程就分析完了。总结一下：

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类似）端的处理器链的交互过程