原创

Storm DRPC实现机制分析剖析

        DRPC是建立在Storm基本概念(Topology、Spout、Bolt、Stream等)之上的高层抽象,个人理解它的目标是在Storm 集群之上提供一种分布式的RPC框架,以便能够利用Storm快速的实现RPC请求的分布式计算过程,即发起一次RPC请求,多个worker计算节点参 与计算,最后汇总后将计算结果返回给客户端。


Storm中使用Thrift作为其RPC框架,同样地,DRPC的实现也是构建在Thrift协议之上,相关的源码文件如下:

1. storm-core/src/storm.thrift,定义了Storm中实现的Thrift协议,其中有两个service是与DRPC相关的:DistributedRPC和DistributedRPCInvocations,它们的接口定义如下:

DistributedRPC.Iface:定义了execute方法,用于客户端发起RPC请求;

DistributedRPCInvocations.Iface:定义了fetchRequest、failRequest、result方法,分别用于获取RPC请求、将RPC请求标记为失败、返回RPC请求的处理结果。


2. storm-core/src/clj/backtype/storm/daemon/drpc.clj,实现了DRPC的Thrift服务端(即DRPC Server),使用Clojure语言实现。


3. storm-core/src/jvm/backtype/storm/generated/DistributedRPC.java 和storm-core/src/jvm/backtype/storm/utils/DRPCClient.java,作为RPC客户端,实现了 DistributedRPC.Iface接口,用于客户端向DRPC Server发起RPC请求。


4. storm-core/src/jvm/backtype/storm/generated /DistributedRPCInvocations.java和storm-core/src/jvm/backtype/storm/drpc /DRPCInvocationsClient.java,作为RPC客户端,实现了DistributedRPCInvocations.Iface接 口,用于DRPC Topology触发执行DRPC Request并返回结果给DRPC Server。

从中可以看出,对于DRPC Server来说,DRPC Client和DRPC Topology都是Thrift的客户端,只是分别调用了不同的Thrift服务而已。


Storm DRPC实现架构

1. 首先,前提是集群上已经运行了DRPC Topology,每个DRPC服务注册了一个RPC方法,包含方法名称和参数形式(上图中假设Topology已经启动运行);


2. 接下来是处理流程,客户端通过DRPCClient调用execute方法,发起一次RPC调用给DRPC Server,目前受限的是只支持一个String类型的DRPC方法调用参数,社区中正在讨论对此进行扩展;


3. 然后,DRPC Server中有一个handler-server pool,用于接收RPC请求,并为每个请求生成唯一的request id,生成一条DRPC Request记录,并放到request queue中等待被消费(计算);


4. 最后,DRPC Topology中的相关模块(DRPC Spout、ReturnResults Bolt,后面会介绍)通过invoke-server pool从request queue中取出该方法的RPC请求,并将处理结果(成功/失败)返回给DRPC Server,直到最终返回给阻塞着的DRPC Client。


Storm DRPC拓扑数据流

其 中,DRPC Topology由1个DRPCSpout、1个Prepare-Request Bolt、若干个User Bolts(即用户通过LinearDRPCTopologyBuilder添加的Bolts)、1个JoinResult Bolt和1个ReturnResults Bolt组成。除了User Bolts以外,其他的都是由LinearDRPCTopologyBuilder内置添加到Topology中的。接下来,我们从数据流的流动关系来 看,这些Spout和Bolts是如何工作的:


1. DRPCSpout中维护了若干个DRPCInvocationsClient,通过fetchRequest方法从DRPC Server读取需要提交到Topology中计算的RPC请求,然后发射一条数据流给Prepare-Request Bolt:<”args”, ‘”return-info”>,其中args表示RPC请求的参数,而return-info中则包含了发起这次RPC请求的RPC Server信息(host、port、request id),用于后续在ReturnResults Bolt中返回计算结果时使用。


2. Prepare-Request Bolt接收到数据流后,会新生成三条数据流:

<”request”, ”args”>:发给用户定义的User Bolts,提取args后进行DRPC的实际计算过程;

<”request”, ”return-info”>:发给JoinResult Bolt,用于和User Bolts的计算结果做join以后将结果返回给客户端;

<”request”>:在用户自定义Bolts实现了FinishedCallback接口的情况下,作为ID流发给用户定义的最后一级Bolt,用于判断batch是否处理完成。


3. User Bolts按照用户定义的计算逻辑,以及RPC调用的参数args,进行业务计算,并最终输出一条数据流给JoinResult Bolt:<”request”, ”result”>。


4. JoinResult Bolt将上游发来的<”request”, ”return-info”>和<”request”, ”result”>两条数据流做join,然后输出一条新的数据流给ReturnResults Bolt: <”result”, ”return-info”>。


5. ReturnResults Bolt接收到数据流后,从return-info中提取出host、port、request id,根据host和port生成DRPCInvocationsClient对象,并调用result方法将request id及result返回给DRPC Server,如果result方法调用成功,则对tuple进行ack,否则对tuple进行fail,并最终在DRPCSpout中检测到tuple 失败后,调用failRequest方法通知DRPC Server该RPC请求执行失败。


参考资料链接

1. https://github.com/apache/incubator-storm

2. https://github.com/nathanmarz/storm-starter

关注下方微信公众号“Java精选”(w_z90110),回复关键字领取资料:如HadoopDubboCAS源码等等,免费领取资料视频和项目。 

涵盖:程序人生、搞笑视频、算法与数据结构、黑客技术与网络安全、前端开发、Java、Python、Redis缓存、Spring源码、各大主流框架、Web开发、大数据技术、Storm、Hadoop、MapReduce、Spark、elasticsearch、单点登录统一认证、分布式框架、集群、安卓开发、iOS开发、C/C++、.NET、Linux、Mysql、Oracle、NoSQL非关系型数据库、运维等。

评论

分享:

支付宝

微信