共計 2927 個字符，預計需要花費 8 分鐘才能閱讀完成。
這篇文章給大家介紹如何進行 Storm DRPC 實現機制分析，內容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。
DRPC 是建立在 Storm 基本概念（Topology、Spout、Bolt、Stream 等）之上的高層抽象，個人理解它的目標是在 Storm 集群之上提供一種分布式的 RPC 框架，以便能夠利用 Storm 快速的實現 RPC 請求的分布式計算過程，即發起一次 RPC 請求，多個 worker 計算節點參 與計算，最后匯總后將計算結果返回給客戶端。
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。
從中可以看出，對于 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。
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 請求執行失敗。
關于如何進行 Storm DRPC 實現機制分析就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。
