Storm高級原語（二） -- DRPC詳解

Storm裏面引入DRPC主要是利用storm的實時計算能力來並行化CPU密集型（CPU intensive）的計算任務。DRPC的stormtopology以函數的參數流做爲輸入，而把這些函數調用的返回值做爲topology的輸出流。

DRPC其實不能算是storm自己的一個特性， 它是經過組合storm的原語stream、spout、bolt、 topology而成的一種模式(pattern)。原本應該把DRPC單獨打成一個包的， 可是DRPC實在是太有用了，因此咱們咱們把它和storm捆綁在一塊兒。

概覽

Distributed RPC是由一個」DPRC服務器」協調(storm自帶了一個實現)。DRPC服務器協調：① 接收一個RPC請求  ② 發送請求到storm topology  ③ 從storm topology接收結果  ④ 把結果發回給等待的客戶端。從客戶端的角度來看一個DRPC調用跟一個普通的RPC調用沒有任何區別。好比下面是客戶端如何調用RPC計算「reach」功能（function）的結果，reach方法的參數是: http://twitter.com。

DRPCClient client =  new DRPCClient("drpc-host", 3772);    String result =  client.execute("reach", "http://twitter.com");

DRPC的工做流大體是這樣的（重要☆）：

客戶端給DRPC服務器發送要執行的函數（function）的名字，以及這個函數的參數。實現了這個函數的topology使用DRPCSpout從DRPC服務器接收函數調用流，每一個函數調用被DRPC服務器標記了一個惟一的id。 這個topology而後計算結果，在topology的最後，一個叫作ReturnResults的bolt會鏈接到DRPC服務器，而且把這個調用的結果發送給DRPC服務器(經過那個惟一的id標識)。DRPC服務器用那個惟一id來跟等待的客戶端匹配上，喚醒這個客戶端而且把結果發送給它。

LinearDRPCTopologyBuilder

Storm自帶了一個稱做LinearDRPCTopologyBuilder的topology builder，它把實現DRPC的幾乎全部步驟都自動化了。這些步驟包括：

一、設置spout

二、把結果返回給DRPC服務器

三、給bolt提供有限聚合幾組tuples的能力

來看一個簡單的例子，下面是一個把輸入參數後面添加一個」!」的DRPC topology的實現：

public static class  ExclaimBolt extends BaseBasicBolt {        public void execute(Tuple tuple,  BasicOutputCollector collector) {            String input = tuple.getString(1);            collector.emit(new  Values(tuple.getValue(0), input + "!"));        }        public void  declareOutputFields(OutputFieldsDeclarer declarer) {            declarer.declare(new  Fields("id", "result"));        }    }       public static void  main(String[] args) throws Exception {        LinearDRPCTopologyBuilder builder = new  LinearDRPCTopologyBuilder("exclamation");        builder.addBolt(new ExclaimBolt(), 3);        // ...    }

能夠看出來，咱們須要作的事情很是的少。建立LinearDRPCTopologyBuilder的時候，你須要告訴它你要實現的DRPC函數(DRPC function)的名字。一個DRPC服務器能夠協調不少函數，函數與函數之間靠函數名字來區分。你聲明的第一個bolt會接收一個兩維tuple，tuple的第一個字段是request-id，第二個字段是這個請求的參數。LinearDRPCTopologyBuilder同時要求咱們topology的最後一個bolt發送一個形如[id, result]的二維tuple：第一個field是request-id，第二個field是這個函數的結果。最後全部中間tuple的第一個field必須是request-id。

在這裏例子裏面ExclaimBolt 簡單地在輸入tuple的第二個field後面再添加一個」!」，其他的事情都由LinearDRPCTopologyBuilder幫咱們搞定：鏈接到DRPC服務器，而且把結果發回。

本地模式DRPC

DRPC能夠以本地模式運行，下面就是以本地模式運行上面例子的代碼：

LocalDRPC drpc = new LocalDRPC();    LocalCluster cluster = new  LocalCluster();       cluster.submitTopology("drpc-demo",  conf, builder.createLocalTopology(drpc));       System.out.println("Results for  'hello':" + drpc.execute("exclamation", "hello"));       cluster.shutdown();    drpc.shutdown();

首先你建立一個LocalDRPC對象，這個對象在進程內模擬一個DRPC服務器（這很相似於LocalCluster在進程內模擬一個Storm集羣），而後建立LocalCluster對象在本地模式運行topology。LinearTopologyBuilder有單獨的方法來建立本地的topology和遠程的topology。在本地模式裏面LocalDRPC對象不和任何端口綁定，因此咱們的topology對象須要知道和誰交互，這就是爲何createLocalTopology方法接受一個LocalDRPC對象做爲輸入的緣由。

把topology啓動了以後，你就能夠經過調用LocalDRPC對象的execute來調用RPC方法了。

遠程模式DRPC

在一個真實集羣上面DRPC也是很是簡單的，有三個步驟:

一、啓動DRPC服務器

二、配置DRPC服務器的地址

三、提交DRPCtopology到storm集羣裏面去。

咱們能夠經過「bin/storm drpc」命令來啓動DRPC服務器。

接着，你須要讓你的storm集羣知道你的DRPC服務器的地址。DRPCSpout須要這個地址從而能夠從DRPC服務器來接收函數調用。這個能夠配置在storm.yaml或者經過代碼的方式配置在topology裏面。經過storm.yaml配置是這樣的：

drpc.servers:      - "drpc1.foo.com"      - "drpc2.foo.com"

最後，你經過StormSubmitter對象來提交DRPC topology（這個跟你提交其它topology沒有區別）。若是要以遠程的方式運行上面的例子，用下面的代碼：

StormSubmitter.submitTopology("exclamation-drpc",  conf, builder.createRemoteTopology());

咱們用createRemoteTopology方法來建立運行在真實集羣上的DRPC topology。

更多精彩內容請關注：http://bbs.superwu.cn

關注超人學院微信二維碼：