如何進(jìn)行Storm DRPC實(shí)現(xiàn)機(jī)制分析

這篇文章給大家介紹如何進(jìn)行Storm DRPC實(shí)現(xiàn)機(jī)制分析，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。

        DRPC是建立在Storm基本概念（Topology、Spout、Bolt、Stream等）之上的高層抽象，個(gè)人理解它的目標(biāo)是在Storm 集群之上提供一種分布式的RPC框架，以便能夠利用Storm快速的實(shí)現(xiàn)RPC請(qǐng)求的分布式計(jì)算過程，即發(fā)起一次RPC請(qǐng)求，多個(gè)worker計(jì)算節(jié)點(diǎn)參 與計(jì)算，最后匯總后將計(jì)算結(jié)果返回給客戶端。

Storm中使用Thrift作為其RPC框架，同樣地，DRPC的實(shí)現(xiàn)也是構(gòu)建在Thrift協(xié)議之上，相關(guān)的源碼文件如下：

1. storm-core/src/storm.thrift，定義了Storm中實(shí)現(xiàn)的Thrift協(xié)議，其中有兩個(gè)service是與DRPC相關(guān)的：DistributedRPC和DistributedRPCInvocations，它們的接口定義如下：

DistributedRPC.Iface：定義了execute方法，用于客戶端發(fā)起RPC請(qǐng)求；

DistributedRPCInvocations.Iface：定義了fetchRequest、failRequest、result方法，分別用于獲取RPC請(qǐng)求、將RPC請(qǐng)求標(biāo)記為失敗、返回RPC請(qǐng)求的處理結(jié)果。

2. storm-core/src/clj/backtype/storm/daemon/drpc.clj，實(shí)現(xiàn)了DRPC的Thrift服務(wù)端（即DRPC Server），使用Clojure語言實(shí)現(xiàn)。

3. storm-core/src/jvm/backtype/storm/generated/DistributedRPC.java 和storm-core/src/jvm/backtype/storm/utils/DRPCClient.java，作為RPC客戶端，實(shí)現(xiàn)了 DistributedRPC.Iface接口，用于客戶端向DRPC Server發(fā)起RPC請(qǐng)求。

4. storm-core/src/jvm/backtype/storm/generated /DistributedRPCInvocations.java和storm-core/src/jvm/backtype/storm/drpc /DRPCInvocationsClient.java，作為RPC客戶端，實(shí)現(xiàn)了DistributedRPCInvocations.Iface接 口，用于DRPC Topology觸發(fā)執(zhí)行DRPC Request并返回結(jié)果給DRPC Server。

從中可以看出，對(duì)于DRPC Server來說，DRPC Client和DRPC Topology都是Thrift的客戶端，只是分別調(diào)用了不同的Thrift服務(wù)而已。

Storm DRPC實(shí)現(xiàn)架構(gòu)

1. 首先，前提是集群上已經(jīng)運(yùn)行了DRPC Topology，每個(gè)DRPC服務(wù)注冊(cè)了一個(gè)RPC方法，包含方法名稱和參數(shù)形式（上圖中假設(shè)Topology已經(jīng)啟動(dòng)運(yùn)行）；

2. 接下來是處理流程，客戶端通過DRPCClient調(diào)用execute方法，發(fā)起一次RPC調(diào)用給DRPC Server，目前受限的是只支持一個(gè)String類型的DRPC方法調(diào)用參數(shù)，社區(qū)中正在討論對(duì)此進(jìn)行擴(kuò)展；

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

4. 最后，DRPC Topology中的相關(guān)模塊（DRPC Spout、ReturnResults Bolt，后面會(huì)介紹）通過invoke-server pool從request queue中取出該方法的RPC請(qǐng)求，并將處理結(jié)果（成功/失?。┓祷亟oDRPC Server，直到最終返回給阻塞著的DRPC Client。

Storm DRPC拓?fù)鋽?shù)據(jù)流

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

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

2. Prepare-Request Bolt接收到數(shù)據(jù)流后，會(huì)新生成三條數(shù)據(jù)流：

<”request”, ”args”>：發(fā)給用戶定義的User Bolts，提取args后進(jìn)行DRPC的實(shí)際計(jì)算過程；

<”request”, ”return-info”>：發(fā)給JoinResult Bolt，用于和User Bolts的計(jì)算結(jié)果做join以后將結(jié)果返回給客戶端；

<”request”>：在用戶自定義Bolts實(shí)現(xiàn)了FinishedCallback接口的情況下，作為ID流發(fā)給用戶定義的最后一級(jí)Bolt，用于判斷batch是否處理完成。

3. User Bolts按照用戶定義的計(jì)算邏輯，以及RPC調(diào)用的參數(shù)args，進(jìn)行業(yè)務(wù)計(jì)算，并最終輸出一條數(shù)據(jù)流給JoinResult Bolt：<”request”, ”result”>。

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

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

關(guān)于如何進(jìn)行Storm DRPC實(shí)現(xiàn)機(jī)制分析就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到。