之前寫過一篇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初始化工程中也相應初始化了這兩個重要元件。同時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資訊。