共计 2232 个字符，预计需要花费 6 分钟才能阅读完成。
Storm 的 DRPC 模式的作用是实现从远程调用 storm 集群的计算资源，而不需要连接到集群的某一个节点。OK。那么 Storm 实现 DRPC 主要是使用 LinearDRPCTopologyBuilder 这个类。下面就先来看看一个简单的例子，它的源码在 github 上。
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;
public class BasicDRPCTopology {
public static class ExclaimBolt extends BaseBasicBolt {// 主要需要覆写 execute 方法和 declareoutputfields 方法
@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
String input = tuple.getString(1);
collector.emit(new Values(tuple.getValue(0), input + “!”));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields(“id”, “result”));
}
}
public static void main(String[] args) throws Exception {
LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder(“exclamation”);// 实现 DRPC 模式
builder.addBolt(new ExclaimBolt(), 3);
Config conf = new Config();
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.out.println(“Result for \”” + word + “\”: ” + drpc.execute(“exclamation”, word));
}
cluster.shutdown();
drpc.shutdown();
}
else {
conf.setNumWorkers(3);
StormSubmitter.submitTopology(args[0], conf, builder.createRemoteTopology());
}
}
}
这段代码主要实现的功能是给接收到的每一个输入后面添加一个感叹号。ok，这样就可以编译提交了。
不过在这之前需要先配置 storm 集群的 drpc server 的 ip。如图。主要是下面的 server 的 ip 需要配置好。并且集群的每一个节点的配置文件都需要配置这项参数！
然后即可使用 storm drpc & 命令启动 drpc 模式。（这里的分工是 172.17.150.6 为客户端，其余的 172.17.150.7（.8，.11）为集群的三个节点，.11 是 nimbus 节点。）
OK，那接下来就使用客户端向集群提交 Topology。如图。使用客户端向集群提交名为 exclaim 的 Topology。里面设置的 worker 数为 3。
从下图可以看到两个 supervisor 分别有一个是运行两个 worker，有一个是运行一个 worker。
ok，下面是客户端调用远程资源进行计算的程序。主要是声明 DRPCClient 的 ip 以及端口，以及指定执行的方法名和传入的参数（client.execute(“exclamation”,word)）。
运行结果如下。
OK，整个 DRPC 的过程就是这样。
Storm 进程通信机制分析 http://www.linuxidc.com/Linux/2014-12/110158.htm
Apache Storm 的历史及经验教训 http://www.linuxidc.com/Linux/2014-10/108544.htm
Apache Storm 的详细介绍 ：请点这里
Apache Storm 的下载地址 ：请点这里
