Storm DRPC實現機制分析

DRPC是建立在Storm基本概念（Topology、Spout、Bolt、Stream等）之上的高層抽象，個人理解它的目標是在Storm 集群之上提供一種分布式的RPC框架，以便能夠利用Storm快速的實現RPC請求的分布式計算過程，即發起一次RPC請求，多個worker計算節點參與計算，最后匯總后將計算結果返回給客戶端。

DRPC實現框架

Storm中使用Thrift作為其RPC框架，同樣地，DRPC的實現也是構建在Thrift協議之上，相關的源碼文件如下：

1. storm-core/src/storm.thrift，定義了Storm中實現的Thrift協議，其中有兩個service是與DRPC相關的：DistributedRPC和DistributedRPCInvocations，它們的接口定義如下：

DistributedRPC.Iface：定義了execute方法，用于客戶端發起RPC請求；
DistributedRPCInvocations.Iface：定義了fetchRequest、failRequest、result方法，分別用于獲取RPC請求、將RPC請求標記為失敗、返回RPC請求的處理結果。

2. storm-core/src/clj/backtype/storm/daemon/drpc.clj，實現了DRPC的Thrift服務端（即DRPC Server），使用Clojure語言實現。

3. storm-core/src/jvm/backtype/storm/generated/DistributedRPC.java 和storm-core/src/jvm/backtype/storm/utils/DRPCClient.java，作為RPC客戶端，實現了 DistributedRPC.Iface接口，用于客戶端向DRPC Server發起RPC請求。

4. storm-core/src/jvm/backtype/storm/generated /DistributedRPCInvocations.java和storm-core/src/jvm/backtype/storm/drpc /DRPCInvocationsClient.java，作為RPC客戶端，實現了DistributedRPCInvocations.Iface接口，用于DRPC Topology觸發執行DRPC Request并返回結果給DRPC Server。

下圖所示是Storm DRPC的整體實現情況。從中可以看出，對于DRPC Server來說，DRPC Client和DRPC Topology都是Thrift的客戶端，只是分別調用了不同的Thrift服務而已。

Storm DRPC實現架構

1. 首先，前提是集群上已經運行了DRPC Topology，每個DRPC服務注冊了一個RPC方法，包含方法名稱和參數形式（上圖中假設Topology已經啟動運行）；

2. 接下來是處理流程，客戶端通過DRPCClient調用execute方法，發起一次RPC調用給DRPC Server，目前受限的是只支持一個String類型的DRPC方法調用參數，社區中正在討論對此進行擴展；

3. 然后，DRPC Server中有一個handler-server pool，用于接收RPC請求，并為每個請求生成唯一的request id，生成一條DRPC Request記錄，并放到request queue中等待被消費（計算）；

4. 最后，DRPC Topology中的相關模塊（DRPC Spout、ReturnResults Bolt，后面會介紹）通過invoke-server pool從request queue中取出該方法的RPC請求，并將處理結果（成功/失敗）返回給DRPC Server，直到最終返回給阻塞著的DRPC Client。

DRPC拓撲實現細節

下面以storm-starter項目中的ReachTopology為例，詳細介紹Storm中通過LinearDRPCTopologyBuilder構造的DRPC Topology的數據流關系：

Storm DRPC拓撲數據流

其中，DRPC Topology由1個DRPCSpout、1個Prepare-Request Bolt、若干個User Bolts（即用戶通過LinearDRPCTopologyBuilder添加的Bolts）、1個JoinResult Bolt和1個ReturnResults Bolt組成。除了User Bolts以外，其他的都是由LinearDRPCTopologyBuilder內置添加到Topology中的。接下來，我們從數據流的流動關系來看，這些Spout和Bolts是如何工作的：

1. DRPCSpout中維護了若干個DRPCInvocationsClient，通過fetchRequest方法從DRPC Server讀取需要提交到Topology中計算的RPC請求，然后發射一條數據流給Prepare-Request Bolt：<”args”, ‘”return-info”>，其中args表示RPC請求的參數，而return-info中則包含了發起這次RPC請求的RPC Server信息（host、port、request id），用于后續在ReturnResults Bolt中返回計算結果時使用。

2. Prepare-Request Bolt接收到數據流后，會新生成三條數據流：

<”request”, ”args”>：發給用戶定義的User Bolts，提取args后進行DRPC的實際計算過程；
<”request”, ”return-info”>：發給JoinResult Bolt，用于和User Bolts的計算結果做join以后將結果返回給客戶端；
<”request”>：在用戶自定義Bolts實現了FinishedCallback接口的情況下，作為ID流發給用戶定義的最后一級Bolt，用于判斷batch是否處理完成。

3. User Bolts按照用戶定義的計算邏輯，以及RPC調用的參數args，進行業務計算，并最終輸出一條數據流給JoinResult Bolt：<”request”, ”result”>。

4. JoinResult Bolt將上游發來的<”request”, ”return-info”>和<”request”, ”result”>兩條數據流做join，然后輸出一條新的數據流給ReturnResults Bolt： <”result”, ”return-info”>。

5. ReturnResults Bolt接收到數據流后，從return-info中提取出host、port、request id，根據host和port生成DRPCInvocationsClient對象，并調用result方法將request id及result返回給DRPC Server，如果result方法調用成功，則對tuple進行ack，否則對tuple進行fail，并最終在DRPCSpout中檢測到tuple 失敗后，調用failRequest方法通知DRPC Server該RPC請求執行失敗。