欢迎转载,请注明出处:http://blog.csdn.net/jmppok/article/details/16839363
1. DRPC介绍
Storm是一个分布式实时处理框架,它支持以DRPC方式调用.可以理解为Storm是一个集群,DRPC提供了集群中处理功能的访问接口.
其实即使不通过DRPC,而是通过在Topoloye中的spout中建立一个TCP/HTTP监听来接收数据,在最后一个Bolt中将数据发送到指定位置也是可以的。这是后话,后面再进行介绍。而DPRC则是Storm提供的一套开发组建,使用DRPC可以极大的简化这一过程。
2.DRPC的使用
DRPC包括服务端和客户端两部分。引用官方的一张图片来进行说明:
1)服务端
服务端由四部分组成:包括一个DRPC Server, 一个 DPRC Spout,一个Topology和一个ReturnResult。
在实际使用中,主要有三个步骤:
a.启动Storm中的DRPC Server;
首先,修改Storm/conf/storm.yaml中的drpc server地址;需要注意的是:必须修改所有Nimbus和supervisor上的配置文件,设置drpc server地址。否则在运行过程中可能无法返回结果。
然后,通过 storm drpc命令启动drpc server。
b.创建一个DRPC 的Topology,提交到storm中运行。
该Toplogy和普通的Topology稍有不同,可以通过两种方式创建:
创建方法一:直接使用 Storm 提供的LinearDRPCTopologyBuilder。 (不过该方法在0.82版本中显示为已过期,不建议使用)
LinearDRPCTopologyBuilder 可以很方便的创建一个DRPC 的Topology,经过本人测试可以运行,但在使用C++开发的Bolt时有异常,具体问题见:http://blog.csdn.net/jmppok/article/details/16827837
1 | 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); 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.err.println("Result for \"" + word + "\": "+ drpc.execute("exclamation", word));StormSubmitter.submitTopology("exclamation", conf,builder.createRemoteTopology()); |
创建方法二:直接使用 Storm 提供的通用TopologyBuilder。 不过需要自己手动加上开始的DRPCSpout和结束的ReturnResults。
其实Storm 提供的LinearDRPCTopologyBuilder也是通过这种封装而来的。
1 | TopologyBuilder builder = new TopologyBuilder(); DRPCSpout drpcSpout = new DRPCSpout("exclamation"); builder.setSpout("drpc-input", drpcSpout,5); builder.setBolt("cpp", new CppBolt(), 5) .noneGrouping("drpc-input"); builder.setBolt("return", new ReturnResults(),5) Config conf = new Config(); conf.setMaxTaskParallelism(3); StormSubmitter.submitTopology("exclamation", conf,builder.createTopology()); |
c.通过DRPCClient对Cluster进行访问
需要修改客户端配置文件 ~/.storm/storm.yaml,配置drpc server的地址。修改方法可storm服务端一样。
访问代码就很简单了:
1 | DRPCClient client = new DRPCClient("10.100.211.232", 3772);String result = client.execute("exclamation","test"); |
注意如果是本地模式,topology的提交和drpc的访问都有不同。
1 | 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)客户端
客户端在上面服务端中已经介绍了