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

DRPC实现框架

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。

下图所示是Storm DRPC的整体实现情况。从中可以看出,对于DRPC Server来说,DRPC Client和DRPC Topology都是Thrift的客户端,只是分别调用了不同的Thrift服务而已。

Storm DRPC实现机制分析-LMLPHP

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。

DRPC拓扑实现细节

下面以storm-starter项目中的ReachTopology为例,详细介绍Storm中通过LinearDRPCTopologyBuilder构造的DRPC Topology的数据流关系:

Storm DRPC实现机制分析-LMLPHP

其中,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请求执行失败。

DRPC整体工作流程

最后,将以上两个小节的内容合并起来就可以看到Storm中DRPC的整体工作流程了,如下图所示:

Storm DRPC实现机制分析-LMLPHP

参考资料链接

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

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

04-16 07:33