Storm同步调用之DRPC模型探讨

本文涉及的产品
云数据库 Tair(兼容Redis),内存型 2GB
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
简介:   摘要:Storm的编程模型是一个有向无环图,决定了storm的spout接收到外部系统的请求后,spout并不能得到bolt的处理结果并将结果返回给外部请求。所以也就决定了storm无法提供对外部系统的同步调用功能。

  摘要:Storm的编程模型是一个有向无环图,决定了storm的spout接收到外部系统的请求后,spout并不能得到bolt的处理结果并将结果返回给外部请求。所以也就决定了storm无法提供对外部系统的同步调用功能。

 

  最近新的黑名单项目需要在storm实时计算平台上提供对外部系统请求调用的同步响应(也就是让storm支持同步调用而不是回调),而Storm的编程模型是一个有向无环图,也就决定了storm的spout接收到外部系统的请求后,将请求数据分发给下游的bolt进行处理后,spout并不能得到bolt的处理结果并将结果返回给外部请求。

  在传统也就是业界大部分应用场景storm对外部系统的调用都是采用回调的方式。本人之前参与的某4000万用户,日均1000万交易量的信用卡中心也是采用回调的方式。

 

原文和作者一起讨论:http://www.cnblogs.com/intsmaze/p/7602242.html

        

      storm常见回调设计方案

  首先jetty,tomcat等启动服务,接收外部系统的请求,将请求得到的数据发往kafka,activeMQ等消息队列中,就立马响应给外部系统。

  然后storm实时平台去消息队列中拉取数据并进行分布式并行处理,然后将运算完的结果存入第三方存储介质(外部系统直接通过读取该介质获取结果)或者调用外部系统的接口将处理的结果推送出去(以回调的方式实现伪同步请求)。

  

   目前的需求

  现在的项目是一个产品,要接入各大银行的系统中,所以通过要求对方提供一个回调接口来实现同步是不可能的。必须依靠自己去实现同步请求响应,外部系统将消息发往storm实时平台,然后外部系统会阻塞,等待storm实时平台处理完后将结果返回给外部系统。

   这个时候当然就是去storm的官网去看看有没有对应的高级接口,果不其然看到了DRPC,熟悉RPC的就知道就是远程过程调用,就是向远程系统发送socket请求并得到远程系统处理的结果,那么DPRC也就是分布式远程过程调用而已,那么他就一定提供了同步请求响应的功能。

   关于DRPC在文章末尾会简单演示一下,这里重点说下我对storm的DRPC的原理理解。上面我也说了storm的编程模型是一个有向无环图,从模型的角度来说是不可能支持同步请求的功能的。

   自己如何基于storm实现同步调用

   我也自己思考下,如果是我自己会如何在现有的storm的编程模型下如何实现同步调用。
    
   方案一:大家最容易想到的方案就是,在storm的拓扑的spout节点中new ServerSocket(8080),来接收外部系统的请求,然后将请求的数据分发给下游的bolt处理,处理完后将结果返回给外部系统。 

  问题一:storm的计算模型的拓扑结构是一个有向无环图,处理的结果并不会返回给spout节点。

  我可以让bolt将处理的结果存入redis,然后spout不断轮询去redis读取对应的结果并返回!

  貌似可以,但是查看spout的调用源代码会发现,如果这样会导致spout的吞吐量下降,因为spout只有从redis轮询到当次请求的处理结果后才会在循环调用nextTuple()方法,当然在spout实现类中开启多线程后,貌似可以解决nextTuple方法阻塞(具体没有去想,因为本身这个方案不可行了,就没必须去掉头发了)storm的任务中再去开多线程是无效率的,还不如不选择storm技术。

  问题二:spout节点启动的机器是不固定的,ip是会变化的,则对外部系统调用时ip的维护带来了麻烦,所以这种方案不可取。

public void nextTuple() {
        获取请求的数据
        collector.emit();
        while(true)
        {
            去redis中读取该次请求的结果,读到则结束循环
        }    
 }

   

  方案二:抛开storm实时平台,单独开发一套中转程序,负责接收外部系统的请求,将外部请求的参数存入一个先进先出的队列中,阻塞等待storm处理的结果。storm拓扑的spout中创建socket去连接中转程序,中转程序从队列中拿出请求参数返回给spout。spout获取到请求参数后,将参数传给下游的bolt去计算,下游的最后一层bolt计算完也创建socke去连接中转程序并将结果发送给中转程序。中转程序获得bolt返回结果,存入某个地方,然后中转程序中阻塞的地方轮询得到结果后,就结束轮询响应给外部系统了。

 

  当然这只是一个简单的方案设计,具体还有很多细节设计以及考虑在我们的Server端,因为它要同时协调三个不同的程序的请求,并且能够根据以每一个请求自动聚合外部系统请求,spout请求,bolt请求为一组。

  Storm的DRPC概述

  storm的DRPC其实就实现外部系统同步调用storm实时平台的功能组件了。应该不需要我去从零开发了。接下来就看看storm的DPRC功能是否和我当初的想法是否一致!

  官方话语:

  分布式RPC(DRPC)背后的思想是将真正强大功能的计算与storm的计算并行化。Storm拓扑以一个函数参数的流作为输入,它向每个函数调用发出一个输出流的结果。

  分布式RPC(DRPC)的真正目的是使用storm实时并行计算极端功能。Storm拓扑需要一个输入流作为函数参数,以一个输出流的形式发射每个函数调用的结果。。从一个客户端的角度来看,一个分布式RPC调用就像是一个常规的RPC调用。

  分布式RPC工作流程如下图所示:

 

  客户端程序会向启动的DRPC服务器发送要执行的函数名称和该函数的参数。具备DRPC功能的拓扑会使用一个DRPCSpout接收来自DRPC服务器传来的函数调用流。每个函数调用都用一个惟一的id标记在DRPC服务器上。拓扑计算好结果后会由一个名为ReturnResults的bolt去连接DRPC服务器给出对应函数调用id的结果,然后DRPC服务器根据ID找到等待中的客户端,为等待中的客户端消除阻塞,并发送结果给客户端。

  从一个客户端的角度来看,一个分布式RPC调用就像是一个常规的RPC调用。

public class Client {
    public static void main(String[] args) throws TException,
            DRPCExecutionException {
        DRPCClient client = new DRPCClient("192.168.19.131", 3772);
        for (int i = 0; i < 10; i++) {
            System.out.println(i);
             String result = client.execute("method_name","param is intsmaze--"+i+"---");
            System.out.println(result);
        }
        client.close();
    }
}

   下一篇将会重点讲解如何运行storm的drpc示例,并剖析它的内部实现原理来验证是否和本文的猜想一致。

作者: intsmaze(刘洋)
老铁,你的--->推荐,--->关注,--->评论--->是我继续写作的动力。
微信公众号号:Apache技术研究院
由于博主能力有限,文中可能存在描述不正确,欢迎指正、补充!
本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则保留追究法律责任的权利。
相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
6月前
|
存储 监控 API
Flink的每个key状态 和每个并行度的状态 怎么测试区分?
Flink的每个key状态 和每个并行度的状态 怎么测试区分?
60 0
|
消息中间件 缓存 Kafka
连Producer端的主线程模块运行原理都不清楚,就敢说自己精通Kafka?
连Producer端的主线程模块运行原理都不清楚,就敢说自己精通Kafka?
58 0
|
3月前
|
消息中间件 监控 Kafka
实时计算 Flink版产品使用问题之处理Kafka数据顺序时,怎么确保事件的顺序性
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
6月前
|
消息中间件 Oracle 关系型数据库
实时计算 Flink版操作报错之连接外部kafka本地执行测试代码报错如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
流计算
124 Storm 任务提交的过程
124 Storm 任务提交的过程
32 0
|
存储 缓存 分布式计算
Flink / Scala - ProcessFunction 之间共用缓存测试
Flink 开发中有如下场景,数据需要经过两次 ProcessFunction 处理,第一步 ProcessV1的一些信息重复不想通过每条数据传输至 ProcessV2,这时便捷的方法时对 ProcessV1 需要存储的元素进行去重缓存,保证全局共用一份缓存,可以有效减少储存空间,下面分别尝试三种缓存方式: A.ValueState 缓存 B.HashMapCache 缓存 C.RedisCahce 缓存...
311 0
Flink / Scala - ProcessFunction 之间共用缓存测试
|
消息中间件 存储 缓存
图解Kafka消息发送者核心参数与工作机制
图解Kafka消息发送者核心参数与工作机制
图解Kafka消息发送者核心参数与工作机制
|
存储 消息中间件 传感器
超越Storm,SparkStreaming——Flink如何实现有状态的计算
超越Storm,SparkStreaming——Flink如何实现有状态的计算
194 0
超越Storm,SparkStreaming——Flink如何实现有状态的计算
|
人工智能 算法 流计算
Flink-CEP论文与源码解读之状态与状态转换
Flink CEP的论文与设计 Flink的CEP设计与实现重度参考了论文《Efficient Pattern Matching over Event Streams》。下面我们就来结合论文谈谈Flink CEP的设计。
2922 0