Storm版本0.9.5
在storm中DRPC服务应用于远程分布式计算，根据客户端提交的请求参数，而返回Storm计算的结果。
DRPC服务启动流程（远程模式）
- 启动DRPC服务，启动命令:storm drpc
- 修改storm.yaml配置文件，添加drpc.servers，如：
drpc.servers:
- "mast"
- 向storm集群中的drpc服务节点mast提交DRPC拓扑
DRPC工作流程
- 客户端发起请求到DRPC Service
- DRPC Service会为这次请求生成对应的requestId与输入的参数一起发送到DRPCSpout，再提交到DrpcTopology(输入参数的字段名称固定为args)
- 处理后的结果返回给ReturnResults
- 调用DRPC Service服务返回给客户端,返回的结果包括requestId和请求的结果数据.
- ReturnResults输出两个字段：第一个为结果数据，第二个为requestId
工作流程图
本地调用DRPC服务方式
| 1 | TopologyBuilder topologyBuilder = new TopologyBuilder(); |
| 2 | LocalDRPC drpc = new LocalDRPC(); |
| 3 | |
| 4 | DRPCSpout spout = new DRPCSpout("test", drpc); |
| 5 | topologyBuilder.setSpout("drpc", spout); |
| 6 | topologyBuilder.setBolt("bolt1", new TestBolt(),2).shuffleGrouping("drpc"); |
| 7 | topologyBuilder.setBolt("return", new ReturnResults(),2).shuffleGrouping("bolt1"); |
| 8 | |
| 9 | LocalCluster cluster = new LocalCluster(); |
| 10 | Config conf = new Config(); |
| 11 | cluster.submitTopology("bolt1", conf, topologyBuilder.createTopology()); |
分布式集群模式调用DRPC服务
| 1 | TopologyBuilder topologyBuilder = new TopologyBuilder(); |
| 2 | |
| 3 | DRPCSpout spout = new DRPCSpout("test"); |
| 4 | topologyBuilder.setSpout("drpc", spout); |
| 5 | topologyBuilder.setBolt("bolt1", new TestBolt(),2).shuffleGrouping("drpc"); |
| 6 | topologyBuilder.setBolt("return", new ReturnResults(),2).shuffleGrouping("bolt1"); |
| 7 | |
| 8 | Config conf = new Config(); |
| 9 | StormSubmitter.submitTopology("bolt1", conf, topologyBuilder.createTopology()); |
调用
| 1 | public class DRPCTest { |
| 2 | |
| 3 | public static void main(String[] args) { |
| 4 | |
| 5 | DRPCClient client = new DRPCClient("远程服务IP地址", 3772); |
| 6 | try { |
| 7 | String result = client.execute("exclamation", "hello "); |
| 8 | |
| 9 | System.out.println(result); |
| 10 | } catch (TException e) { |
| 11 | e.printStackTrace(); |
| 12 | } catch (DRPCExecutionException e) { |
| 13 | e.printStackTrace(); |
| 14 | } |
| 15 | } |
| 16 | } |