【Storm】DRPC精解和案例分析

一、DRPC简介和工作流程

1、 DRPC 简介

分布式RPC( distributed RPC,DRPC) 用于对storm上大量的函数调用进行并行计算。对于每一次函数调用，storm集群上运行的拓扑接收调用函数的参数信息作为输入流，并将计算结果作为输出流发射出去。

一句话概括：storm进行计算，根据客户端提交的请求参数，而返回storm计算的结果。

DRPC通过DRPC Server来实现，DRPC Server的整体工作过程如下：

接收到一个 RPC 调用请求；

发送请求到storm上的拓扑；

从storm上接收计算结果；

将计算结果返回给客户端；

注：在client客户端来看，一个DRPC调用看起来和普通的RPC调用没什么区别；

2、工作流程

DRPC的工作流程如下

client客户端给DRPC Server发送要被调用执行DRPC函数名称及参数。

storm上的topology通过DRPC Spout实现这一函数，从DRPC Server接收到函数调用流，DRPC Server为每次函数调用生成唯一的id；

storm上运行的Topology开始计算结果，最后通过一个 ReturnResults 的Bolt连接到DRPC服务器，发送指定id的计算结果给DRPC Server。

DRPC Server 通过使用之前为每个函数调用生成的id，将结果关联到对应的发起调用的client，将计算结果返回给client。

3、LinearDRPCTopologyBuilder类

LinearDRPCTopologyBuilder 类是官方封装好的，专门用于DRPC的builder，但是从 storm0.82 已经不推荐使用该类，原因是因为在trident已经有更好的封装；

Storm自带了一个称作 LinearDRPCTopologyBuilder 的topology builder, 它把实现DRPC的几乎所有步骤都自动化了。
包括:
1. 构建spout；
2. 向DRPC Server返回结果；
3. 为bolt提供函数用于对tuple进行处理；

二、官方DRPC入门案例

在官方给出的storm-starter-master工程，storm.starter包，有DRPC案例；

1、BasicDRPCTopology案例

用官方封装好的 LinearDRPCTopologyBuilder 类，比较简单。本地提交。

package storm.starter;import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.drpc.LinearDRPCTopologyBuilder;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;/*** This topology is a basic example of doing distributed RPC on top of Storm. It implements a function that appends a* "!" to any string you send the DRPC function.* <p/>* See https://github.com/nathanmarz/storm/wiki/Distributed-RPC for more information on doing distributed RPC on top of* Storm.*/
public class BasicDRPCTopology {public static class ExclaimBolt extends BaseBasicBolt {@Overridepublic void execute(Tuple tuple, BasicOutputCollector collector) {String input = tuple.getString(1);// 发射的第一列是请求id，第二列是数据。对收到的数据加上 "!" 之后返回到 DRPC Server，collector.emit(new Values(tuple.getValue(0), input + "!"));}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("id", "result"));}}public static void main(String[] args) throws Exception {// LinearDRPCTopologyBuilder 类是官方封装好的，专门用于DRPC的builder。且该builder没有数据源；// 构造函数的入参，是函数名称 "exclamation" ，在storm上必须唯一；LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("exclamation");// 设置bolt的实例为3个builder.addBolt(new ExclaimBolt(), 3);Config conf = new Config();conf.setDebug(false);// 如果没有参数，本地模式提交if (args == null || args.length == 0) {LocalDRPC drpc = new LocalDRPC();LocalCluster cluster = new LocalCluster();cluster.submitTopology("drpc-demo", conf, builder.createLocalTopology(drpc));// 循环，每次调用函数和传参数。这里是客户端代码for (String word : new String[]{ "hello", "goodbye" }) {System.err.println("Result for \"" + word + "\": " + drpc.execute("exclamation", word));}// 运行程序后，会报错，注释这2行就不会报错
//      cluster.shutdown();
//      drpc.shutdown();}else {     // 有参数，分布式提交。分布式提交不用传 drpcconf.setNumWorkers(3);StormSubmitter.submitTopology(args[0], conf, builder.createRemoteTopology());}}
}

运行程序后，控制台打印日志

Result for "hello": hello!
Result for "goodbye": goodbye!

2、ManualDRPC案例

用TopologyBuilder 来实现 DRPC，实现bolt可以个性化，增加功能。本地提交。

package drpc;import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.drpc.DRPCSpout;
import backtype.storm.drpc.ReturnResults;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;public class ManualDRPC {public static class ExclamationBolt extends BaseBasicBolt {@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("result", "return-info"));}@Overridepublic void execute(Tuple tuple, BasicOutputCollector collector) {String arg = tuple.getString(0);Object retInfo = tuple.getValue(1);collector.emit(new Values(arg + "!!!", retInfo));}}public static void main(String[] args) {TopologyBuilder builder = new TopologyBuilder();LocalDRPC drpc = new LocalDRPC();// 构造 DRPCSpoutDRPCSpout spout = new DRPCSpout("exclamation", drpc);builder.setSpout("drpc", spout);builder.setBolt("exclaim", new ExclamationBolt(), 3).shuffleGrouping("drpc");builder.setBolt("return", new ReturnResults(), 3).shuffleGrouping("exclaim");LocalCluster cluster = new LocalCluster();Config conf = new Config();conf.setDebug(false);cluster.submitTopology("exclaim", conf, builder.createTopology());System.out.println(drpc.execute("exclamation", "aaa"));System.out.println(drpc.execute("exclamation", "bbb"));}
}

运行程序后，控制台打印日志

aaa!!!
bbb!!!

三、远程模式DRPC

1、远程DRPC简介

远程模式也就是分布式提交。远程模式的DRPC与本地模式的DRPC不同之处在于：远程模式DRPC不需要模拟DRPC Server，而是通过在真实的Storm集群中配置DRPC Server来完成，远程模式通过调用builder的createRemoteTopology方法来构建topology。

在实际的storm集群上运行DRPC也一样很简单，只需要完成以下步骤：

a、启动DRPC Server(s)；

b、配置DRPC Server(s)地址；

c、向storm集群提交DRPC拓扑。

首先，通过storm脚本启动DRPC Server：

bin/storm drpc

然后，在storm集群中配置DRPC Server地址，这就是DRPCSpout读取函数调用请求的地方，这一步的配置可以通过storm.yaml文件或者拓扑的配置来完成。通过storm.yaml文件的配置方式如下（设置主机地址）：

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

2、准备代码

（1）ManualDRPC，主程序类，包含bolt类。

package drpc;import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.drpc.DRPCSpout;
import backtype.storm.drpc.ReturnResults;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;public class ManualDRPC {public static class ExclamationBolt extends BaseBasicBolt {@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("result", "return-info"));}@Overridepublic void execute(Tuple tuple, BasicOutputCollector collector) {String arg = tuple.getString(0);Object retInfo = tuple.getValue(1);// 第一列是数据，在传进来的字符串后面加上3个感叹号collector.emit(new Values(arg + "!!!", retInfo));}}public static void main(String[] args) {TopologyBuilder builder = new TopologyBuilder();LocalDRPC drpc = new LocalDRPC();if (args.length > 0) {// 构造 DRPCSpout，分布式模式不需要第二个参数drpcDRPCSpout spout = new DRPCSpout("exclamation");builder.setSpout("drpc", spout);builder.setBolt("exclaim", new ExclamationBolt(), 3).shuffleGrouping("drpc");builder.setBolt("return", new ReturnResults(), 3).shuffleGrouping("exclaim");Config conf = new Config();conf.setDebug(false);try {StormSubmitter.submitTopology(args[0], conf, builder.createTopology());} catch (AlreadyAliveException | InvalidTopologyException e) {e.printStackTrace();} catch (Exception e) {e.printStackTrace();}} else {// 构造 DRPCSpout，本地模式需要第二个参数drpcDRPCSpout spout = new DRPCSpout("exclamation", drpc);builder.setSpout("drpc", spout);builder.setBolt("exclaim", new ExclamationBolt(), 3).shuffleGrouping("drpc");builder.setBolt("return", new ReturnResults(), 3).shuffleGrouping("exclaim");LocalCluster cluster = new LocalCluster();Config conf = new Config();conf.setDebug(false);cluster.submitTopology("exclaim", conf, builder.createTopology());}// 上传到storm集群做分布式提交，下面两行要注释
//    System.out.println(drpc.execute("exclamation", "aaa"));
//    System.out.println(drpc.execute("exclamation", "bbb"));}
}

（2）MyDRPCClient，客户端类

package drpc;import org.apache.thrift7.TException;import backtype.storm.generated.DRPCExecutionException;
import backtype.storm.utils.DRPCClient;public class MyDRPCClient {public static void main(String[] args) {// storm 集群的master主机地址DRPCClient client = new DRPCClient("192.168.178.131", 3772);try {String result = client.execute("exclamation", "hello");System.out.println(result);} catch (TException | DRPCExecutionException e) {e.printStackTrace();}}
}

（3）现在本地自己，运行ManualDRPC类，用的是本地模式；没问题再打包。

3、打包发布到storm集群

（1）项目右键，Run As -> maven install，生成安装包，在工程target目录下有2个jar包，长名称的包含依赖，这个才是可用的。

（2）用filezilla，把 storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jar 包传送到 hadoop-senior 主机的指定目录（因人而异，不影响结果） /opt/datas/stormjars

4、配置DRPC

在 /opt/stormmodules/storm-0.9.0.1/conf 目录下的 storm.yaml 加上以下内容（主机地址可变）

 drpc.servers:- "hadoop-senior.ibeifeng.com"- "hadoop-senior02.ibeifeng.com"

5、启动storm集群

首先启动zookeeper，在集群的每一台主机的 zookeeper 安装的 bin 目录下运行命令

./zkServer.sh start

启动storm集群，storm的master主机，storm安装的bin目录下运行2个命令，第二个命令是启动监控界面

nohup ./storm nimbus &
nohup ./storm ui &

storm的slave主机，storm安装的bin目录下运行命令

nohup ./storm supervisor &

打开监控界面 http://hadoop-senior:8081/ ，可以看到集群已经启动

6、启动 DRPC 服务

在集群的每一台主机的 storm安装 bin 目录下运行命令：

nohup storm drpc &

可以在监控界面看到drpc已经启动

7、提交拓扑任务

进入上文保存storm案例jar包的目录 /opt/datas/stormjars ，运行命令，后面2个参数依次表示类名 drpc.ManualDRPC，拓扑任务名为 testDRPC。类名必须包含包名，不然寻找不到具体的类。

storm jar ./storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jar drpc.ManualDRPC testDRPC

可以在监控界面看到拓扑任务已经启动

8、从本地发起调用

从上文的 MyDRPCClient 客户端类，运行代码，发起调用，

package drpc;import org.apache.thrift7.TException;import backtype.storm.generated.DRPCExecutionException;
import backtype.storm.utils.DRPCClient;public class MyDRPCClient {public static void main(String[] args) {// storm 集群的master主机地址DRPCClient client = new DRPCClient("192.168.178.131", 3772);try {String result = client.execute("exclamation", "hello");System.out.println(result);} catch (TException | DRPCExecutionException e) {e.printStackTrace();}}
}

入参数是 "hello"，可以看到控制台打印出 "hello!!!" ，说明调用拓扑任务成功

hello!!!