在Storm集群上开启DRPC功能，
基于Storm的1.0.1版本，
并且执行简单的例子测试。

1.DRPC概念

DRPC就是分布式远程过程调用。
Storm里面引入DRPC主要是利用storm的实时计算能力，
来并行化CPU intensive的计算。
DRPC的Storm topology以函数的参数流作为输入，
而把这些函数调用的返回值作为topology的输出流。
DRPC其实不能算是Storm本身的一个特性，
它是通过组合Storm的原语spout,bolt,topology而成的一种模式(pattern)。

2.DRPC工作机制

Distributed RPC是由一个DPRC Server协调的,
Storm自带了一个称作LinearDRPCTopologyBuilder的topology builder，
它把实现DRPC的几乎所有步骤都自动化了。

DRPC服务器协调机制：

接收一个RPC请求；
发送请求到Storm topology；
从Storm topology接收结果；
把结果发回给等待的客户端。

从客户端的角度来看一个DRPC调用，
跟一个普通的RPC调用没有任何区别。
下面是客户端代码展示了如何调用RPC的
exclaimation方法,方法的参数是hello：

DRPCClient client = new DRPCClient("drpc-host", 3772);
String result = client.execute("exclaimation", "hello");

DRPC的工作流大致是这样的:

客户端给DRPC服务器发送要执行的方法的名字，
以及这个方法的参数。
实现了这个函数的topology使用DRPCSpout从DRPC服务器接收函数调用流。
每个函数调用被DRPC服务器标记了一个唯一的id。
这个topology然后计算结果，
在topology的最后一个叫做ReturnResults的bolt会连接到DRPC服务器，
并且把这个调用的结果发送给DRPC服务器(通过那个唯一的id标识)。
DRPC服务器用那个唯一id来跟等待的客户端匹配上，
唤醒这个客户端并且把结果发送给它。

3.配置DPRC Server

修改storm.yaml文件，增加drpc的配置：

drpc.servers:  - "zdh-237"
drpc.childopts: "-Xmx1024m"

其他参数drpc.port, drpc.http.port等使用默认值即可，
参考默认值如下：

drpc.port:3772
drpc.worker.threads:64
drpc.max_buffer_size:1048576
drpc.queue.size:128
drpc.invocations.port:3773
drpc.invocations.threads:64
drpc.request.timeout.secs:600
drpc.childopts:
"-Xmx768m"
drpc.http.port:3774
drpc.https.port:-1
drpc.https.keystore.password:
""
drpc.https.keystore.type:
"JKS"
drpc.http.creds.plugin:backtype.storm.security.auth.DefaultHttpCredentialsPlugin
drpc.authorizer.acl.filename:
"drpc-auth-acl.yaml"
drpc.authorizer.acl.strict:
false

4.启动DPRC Server

使用如下命令启动DRPC进程：
storm drpc > drpc.log 2>&1 &

5.使用本地集群测试

由于命令无入参即没有topo名字，
无对外端口无法提供客户端调用，
在BasicDRPCTopology中启动后调用本地集群，
仅作为测试场景使用。

进入Storm目录，提交处理drpc的topo：

cd /home/stormna/apache-storm-1.0.1/examples/storm-starter/
storm jar storm-starter-topologies-1.0.1.jar org.apache.storm.starter.BasicDRPCTopology

在输出的日志中可以看到如下结果,
结果是单词后面被添加了感叹号**!**

8043 [Thread-26-bolt2-executor[6 6]] INFO  o.a.s.l.ThriftAccessLogger - Request ID: 3 access from:  principal:  operation: result
Result for "hello": hello!
8054 [Thread-26-bolt2-executor[6 6]] INFO  o.a.s.l.ThriftAccessLogger - Request ID: 3 access from:  principal:  operation: result
Result for "goodbye": goodbye!

6.使用真实集群测试

基于真实集群的DRPC可以提供给外部客户端调用，
先提交处理drpc的topo,指定topo名称为exclamationDrpc：

cd /home/stormna/apache-storm-1.0.1/examples/storm-starter
storm jar storm-starter-topologies-1.0.1.jar org.apache.storm.starter.BasicDRPCTopology exclamationDrpc

7.客户端调用

在BasicDRPCTopology中提供的drpc方法名为exclamation,
效果返回结果是在单词后面被添加的感叹号。
使用下面写客户端代码进行调用测试。

7.1.适配storm-core-0.9.6.jar的客户端代码

注意exclamation名称不要拼错，
否则客户端会一直没有响应:

public class DRPCClientTest096 {public static void main(String[] args) throws Exception {String drpcHost = "10.43.159.237";int drpcPort = 3772;DRPCClient client = new DRPCClient(drpcHost, drpcPort);String input="hello1";String result = client.execute("exclamation",input );System.out.println("input:"+input+", result:"+result);}
}

执行DRPCClientTest096类中的main方法，
调用drpc的exclamation函数，传入参数hello1：
控制台输出结果：

input:hello1, result:hello1!

7.2.适配storm-core-1.0.1.jar的客户端代码

注意调用需要配置Storm参数，
和上面的有点区别的。

public class DRPCClientTest101 {public static void main(String[] args) throws Exception {String drpcHost = "10.43.159.237";int drpcPort = 3772;Properties pro = new Properties();// InputStream inStream = new FileInputStream("stormclient.conf");// 读取storm-core-1.0.1.jar里面 的defaults.yaml配置文件InputStream inStream = ClassLoader.getSystemResourceAsStream("defaults.yaml");pro.load(inStream);inStream.close();//由于Properties加载的值带了引号，需要重新设置一下,或者手动去掉前后的引号pro.setProperty("storm.thrift.transport", "org.apache.storm.security.auth.SimpleTransportPlugin");DRPCClient client = new DRPCClient(pro, drpcHost, drpcPort);String input = "hello2";String result = client.execute("exclamation", input);System.out.println("input:" + input + ", result:" + result);}
}

执行DRPCClientTest101类中的main方法，
调用drpc的exclamation函数，传入参数hello2：
控制台输出结果：

input:hello2, result:hello2!

8.参考文章

StormDRPC 概念以及简单例子测试
storm DRPC问题
Storm高级原语（二） — DRPC
storm自带例子详解（二）——BasicDRPCTopology

Storm集群使用DRPC功能Version1.0.1相关推荐

Storm集群安装Version1.0.1开启Kerberos
Storm集群安装,基于版本1.0.1, 同时开启Kerberos安全认证, 使用apache-storm-1.0.1.tar.gz安装包. 1.安装规划角色规划 IP/机器名安装软件运行进程 ...
Storm集群安装Version1.0.1
Storm集群安装,基于版本1.0.1, 使用apache-storm-1.0.1.tar.gz安装包. 1.安装规划角色规划 IP/机器名安装软件运行进程 nimbus zdh-237 sto ...
storm集群操作指南
storm集群操作指南 @(STORM)[storm, 大数据] storm集群操作指南一storm伪分布式安装一环境准备二安装zookeeper 三安装storm 四运行程序二storm集群 ...
Storm集群开启HA高可用
Storm开启HA高可用,包括Nimbus和UI开启两个及以上的进程. 基于已经安装好的Storm集群,开启关键节点角色的HA高可用. Storm安装请参考Storm集群安装Version1.0.1 ...
storm笔记：storm集群
storm笔记:storm集群 Strom集群结构是有一个主节点(nimbus)和多个工作节点(supervisor)组成的主从结构,主节点通过配置静态指定(还有一种主从结构是在运行时动态选举,比如z ...
大数据技术之_17_Storm学习_Storm 概述+Storm 基础知识+Storm 集群搭建+Storm 常用 API+Storm 分组策略和并发度
大数据技术之_17_Storm学习一 Storm 概述 1.1 离线计算是什么? 1.2 流式计算是什么? 1.3 Storm 是什么? 1.4 Storm 与 Hadoop 的区别 1.5 Sto ...
【Storm】Storm简介及Storm集群的安装部署
1.Storm概述 (1)Storm简介 Storm最早是由BackType公司开发的实时处理系统,底层由Clojure实现.Clojure也是一门基于JVM的高级面向函数式的编程语言.2011年Tw ...
zookeeper+kafka集群部署+storm集群
zookeeper+kafka集群部署+storm集群一.环境安装前准备: 准备三台机器操作系统:centos6.8 jdk:jdk-8u111-linux-x64.gz zookeeper:zo ...
Storm集群安装部署步骤【详细版】
作者: 大圆那些事 | 文章可以转载,请以超链接形式标明文章原始出处和作者信息网址: http://www.cnblogs.com/panfeng412/archive/2012/11/30/how ...

Storm集群使用DRPC功能Version1.0.1