共计 2927 个字符，预计需要花费 8 分钟才能阅读完成。
这篇文章给大家介绍如何进行 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 请求执行失败。
关于如何进行 Storm DRPC 实现机制分析就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。
