zoukankan      html  css  js  c++  java
  • Storm DRPC实现机制分析

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

    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实现架构

    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拓扑数据流

    其中,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整体处理流程

    参考资料链接

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

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

  • 相关阅读:
    how to uninstall devkit
    asp.net中bin目录下的 dll.refresh文件
    查找2个分支的共同父节点
    Three ways to do WCF instance management
    WCF Concurrency (Single, Multiple, and Reentrant) and Throttling
    检查string是否为double
    How to hide TabPage from TabControl
    获取当前系统中的时区
    git svn cygwin_exception
    lodoop打印控制具体解释
  • 原文地址:https://www.cnblogs.com/panfeng412/p/storm-drpc-implementation-mechanism-analysis.html
Copyright © 2011-2022 走看看