之前写过一篇dubbo cluster–架构。由于dubbo逻辑集群的功能主要是在client端。主要側重在client的分析。后来由于工作忙和懒癌。也就没再继续server的叙述了。近期正好在看大众点评的cat源代码。当中也有rpc的模块。就借此专门来分析下rpc server的实现。
网络模型
Cat server基于netty,是典型的reactor模型。
上图是网上找的reactor模型演示样例图。Netty的boss和workder分别映射图中的mainReactor和subReactor。由boss accept nio channel。之后交由worker read, decode以及handle等。Netty的实现和图中稍微有点不一致,在netty中decode和handle是同步的。
须要说明的是netty handler也能够异步处理,netty支持线程池分发handler thread。
还有还有一种方案。由应用程序自身实现延迟队列做异步处理,handler仅仅需将消息(事件)放入队列即可,cat採用的就是这样的方案。Cat是通过decoder解码消息后调用handler将消息插入延迟队列,并没有向netty注冊handler再由netty在decode完成后调用对应handler。
Cat的数据传输对象为MessageTree,MessageCodec对传输消息编码和解码。MessageHanlder将消息放入队列。下图是cat server的静态结构。
cat server静态结构
CatHomeModule就是cat server,它包括两个逻辑模块,一个是reactor,还有一个是延时队列(period)。分别对应上图的左右半边。
MessageConsumer和TcpSocketReceiver均被CatHomeModule依赖,事实上是在receiver初始化project中也对应初始化了这两个重要组件。
同一时候MessageConsumer也是MessageHandler聚合属性,而handler则是receiver的一个内部属性。结构上看起来有点混乱。但事实上module是启动了receiver的初始化,然后receiver在初始化过程中依赖了handler,而handler又依赖了consumer。而Module和consumer之间的依赖关系是一种非常弱的关系。仅仅是为了注冊虚拟机的shutdownhook(消息提交章节会做具体说明)。
MessageTree是经由网络传递的消息报文对象,由MessageCodec进行解码和编码。在服务端被解码生成后作为方法參数被MessageConsumer消费,终于放入MessageQueue等待MessageAnalyzer处理。
MessageQueue被聚合在PeriodTask内,后者是个daemon线程,不断轮询MessageQueue,当有队列里有消息时就调用MessageAnalyzer处理,每一个task仅仅对应一个analyzer。analyzer就是队列的消费者。
一个Period代表一个周期,每一个周期对应一个持续时间(duration)。默认是一小时,且周期是整点时间段,比如1:00-2:00,2:00-3:00,而不是1:01-2:01。每一个周期对应多个task,每条消息对应的也会被拷贝成多分分发给每一个task。
周期由PeriodManager參考周期策略(PeriodStrategy)的结果生成或结束。
介绍完cat server的实体概念后,再从动态层面看下它的初始化过程以及消息的就绪,消费和提交过程。
Server初始化
在静态结构视图里提到了CatHomeModule依赖两个实体。分别为MessageCosumer和TcpSocketReceiver,它们分别承担了reaactor以及延迟队列的功能,由CatHomeModule的initialize和setup阶段被初始化。
首先看下延迟队列的初始化过程。
上节提到策略结果会開始一个新的周期或者结束一个老的周期。
上图的步骤5,策略会基于持续时间。提前開始时间(如果为a)以及延迟结束时间(如果为b)生成一个结果(a和b在cat中均默觉得3分钟)。策略结果如果为正则start新周期,为负则end老周期。为0则不做不论什么动作。Cat默认超过duration*2+b的周期会被清理,对应的新周期也会提前a開始。
再看周期启动周期任务的过程(8-12),周期先回从应用上下文中获取全部的MessageAnalyzer(相似spring的getBeansOfType)。再循环每一个analyzer为每一个analyzer生成一个或多个周期任务(视analyzer#getAnalyzerCount值而定)。
再看下reactor模块的初始化过程。
TcpSocketReciver在Server启动过程中被从上下文中lookup出来,而且运行初始化过程,在初始化的过程中通过依赖注入将MessageConsumer注入了MessageHandler中。对应的也将MessageHandler和MessageCodec注入给自己作为聚合属性。
消息就绪
TcpSocketReciver是个netty server。它监听socket请求,由关联的MessageDecoder解析生成MessageTree。再交由MessageHandler处理。
MessageHandler将MessageTree交由MessageConsumer(延迟队列)消费。consumer基于MessageTree的时间戳找到对应的Period(步骤8),将其放入对应的PeriodTask中。
Cat延时队列不支持主题,所以每条消息会默认被全部task消费。
具体描写叙述下步骤8和步骤10:
1. PeriodManager维护了一个m_periods的list,find的过程就是轮询该列表找出duration包括MessageTree#timeStamp的唯一周期。
2. Period里会保存m_tasks映射。结构是(Analyzer类型全限定名,List)。在上节提过analyzer会对应一到多个task。对这样的反复task select过程会通过MessageTree#domain进行hash取模仅仅选出1个的。之所以要支持analyzer和task一对多的关系模型应该是考虑到有些analyzer处理会比較耗时,对这样的就须要设定多个task,保证处理效率。
消息消费
消费过程相对简单,周期任务通过守护线程不断使用MessageAnalyzer对MessageQueue进行分析。如果MessageQueue#poll数据不为空则对poll出的MessageTree进行业务处理。
每种MessageAnalyzer针对具体应用场景做对应处理。用户也能够自己定义analyzer,仅仅须要被容器感知即可。
消息提交
消息被消费后并不会马上持久化。而是放在内存里(analyzer的map属性)。结构是(duration, (domain, T)),T是个业务维度的报表对象。比如EventReport,HeartbeatReport等。
在周期结束或者jvm shutdown时会触发消息持久化。
1. 上图是正常周期结束的处理过程。之前介绍过PeriodManager会默认停掉duration*2+extraTime时间前的周期,被停的周期会循环周期内的全部task做finish,对应的task调用其聚合属性MessageAnalyzer的doCheckpoint做消息持久化。
2. 静态结构章节里介绍过MessageConsumer和CatHomeModule是非常弱的依赖关系。用来注冊虚拟机的shutdownhook。以下是注冊shutdownhook的代码片段:
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
consumer.doCheckpoint();
}
});
consumer#doCheckpoint会通过PeriodManager#findPeriod查找当前时间点对应的周期,并对该周期进行finish,对应上图步骤9及以后。
3. 须要强调的一点是,MessageAnalyzer对MessageTree并非单纯的做保存。而是基于业务做指标度量。能够把MessageTree想象成度量指标,而MessageAnalyzer则是对指标作分析出报表,doCheckpoint持久化的一般都是报表数据。报表里包括了部分甚至全部MessageTree信息。