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

关注超人学院微信二维码：