Storm系列(十七)DRPC介绍

Storm版本0.9.5

在storm中DRPC服务应用于远程分布式计算，根据客户端提交的请求参数，而返回Storm计算的结果。

DRPC服务启动流程（远程模式）

• 启动DRPC服务，启动命令:storm drpc
• 修改storm.yaml配置文件，添加drpc.servers，如：

  drpc.servers:

  - "mast"

• 向storm集群中的drpc服务节点mast提交DRPC拓扑

DRPC工作流程

1. 客户端发起请求到DRPC Service
2. DRPC Service会为这次请求生成对应的requestId与输入的参数一起发送到DRPCSpout，再提交到DrpcTopology(输入参数的字段名称固定为args)
3. 处理后的结果返回给ReturnResults
4. 调用DRPC Service服务返回给客户端,返回的结果包括requestId和请求的结果数据.
5. 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	}