Storm高级原语(二) -- DRPC详解

Storm里面引入DRPC主要是利用storm的实时计算能力来并行化CPU密集型(CPU intensive)的计算任务。DRPC的stormtopology以函数的参数流作为输入,而把这些函数调用的返回值作为topology的输出流。

DRPC其实不能算是storm本身的一个特性, 它是通过组合storm的原语stream、spout、bolt、 topology而成的一种模式(pattern)。本来应该把DRPC单独打成一个包的, 但是DRPC实在是太有用了,所以我们我们把它和storm捆绑在一起。

概览

Distributed RPC是由一个”DPRC服务器”协调(storm自带了一个实现)。DRPC服务器协调:① 接收一个RPC请求  ② 发送请求到storm topology  ③ 从storm topology接收结果  ④ 把结果发回给等待的客户端。从客户端的角度来看一个DRPC调用跟一个普通的RPC调用没有任何区别。比如下面是客户端如何调用RPC计算“reach”功能(function)的结果,reach方法的参数是: http://twitter.com

  

DRPCClient client =  new DRPCClient("drpc-host", 3772);

String result =  client.execute("reach", "http://twitter.com");

DRPC的工作流大致是这样的(重要☆):

客户端给DRPC服务器发送要执行的函数(function)的名字,以及这个函数的参数。实现了这个函数的topology使用DRPCSpout从DRPC服务器接收函数调用流,每个函数调用被DRPC服务器标记了一个唯一的id。 这个topology然后计算结果,在topology的最后,一个叫做ReturnResults的bolt会连接到DRPC服务器,并且把这个调用的结果发送给DRPC服务器(通过那个唯一的id标识)。DRPC服务器用那个唯一id来跟等待的客户端匹配上,唤醒这个客户端并且把结果发送给它。

LinearDRPCTopologyBuilder

Storm自带了一个称作LinearDRPCTopologyBuilder的topology builder,它把实现DRPC的几乎所有步骤都自动化了。这些步骤包括:

1、设置spout

2、把结果返回给DRPC服务器

3、给bolt提供有限聚合几组tuples的能力

来看一个简单的例子,下面是一个把输入参数后面添加一个”!”的DRPC topology的实现:

  

public static class  ExclaimBolt extends BaseBasicBolt {

public void execute(Tuple tuple,  BasicOutputCollector collector) {

String input = tuple.getString(1);

collector.emit(new  Values(tuple.getValue(0), input + "!"));

}

public void  declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new  Fields("id", "result"));

}

}

public static void  main(String[] args) throws Exception {

LinearDRPCTopologyBuilder builder = new  LinearDRPCTopologyBuilder("exclamation");

builder.addBolt(new ExclaimBolt(), 3);

// ...

}

可以看出来,我们需要做的事情非常的少。创建LinearDRPCTopologyBuilder的时候,你需要告诉它你要实现的DRPC函数(DRPC function)的名字。一个DRPC服务器可以协调很多函数,函数与函数之间靠函数名字来区分。你声明的第一个bolt会接收一个两维tuple,tuple的第一个字段是request-id,第二个字段是这个请求的参数。LinearDRPCTopologyBuilder同时要求我们topology的最后一个bolt发送一个形如[id, result]的二维tuple:第一个field是request-id,第二个field是这个函数的结果。最后所有中间tuple的第一个field必须是request-id。

在这里例子里面ExclaimBolt 简单地在输入tuple的第二个field后面再添加一个”!”,其余的事情都由LinearDRPCTopologyBuilder帮我们搞定:连接到DRPC服务器,并且把结果发回。

本地模式DRPC

DRPC可以以本地模式运行,下面就是以本地模式运行上面例子的代码:

  

LocalDRPC drpc = new LocalDRPC();

LocalCluster cluster = new  LocalCluster();

cluster.submitTopology("drpc-demo",  conf, builder.createLocalTopology(drpc));

System.out.println("Results for  ‘hello‘:" + drpc.execute("exclamation", "hello"));

cluster.shutdown();

drpc.shutdown();

首先你创建一个LocalDRPC对象,这个对象在进程内模拟一个DRPC服务器(这很类似于LocalCluster在进程内模拟一个Storm集群),然后创建LocalCluster对象在本地模式运行topology。LinearTopologyBuilder有单独的方法来创建本地的topology和远程的topology。在本地模式里面LocalDRPC对象不和任何端口绑定,所以我们的topology对象需要知道和谁交互,这就是为什么createLocalTopology方法接受一个LocalDRPC对象作为输入的原因。

把topology启动了之后,你就可以通过调用LocalDRPC对象的execute来调用RPC方法了。

远程模式DRPC

在一个真实集群上面DRPC也是非常简单的,有三个步骤:

1、启动DRPC服务器

2、配置DRPC服务器的地址

3、提交DRPCtopology到storm集群里面去。

我们可以通过“bin/storm drpc”命令来启动DRPC服务器。

接着,你需要让你的storm集群知道你的DRPC服务器的地址。DRPCSpout需要这个地址从而可以从DRPC服务器来接收函数调用。这个可以配置在storm.yaml或者通过代码的方式配置在topology里面。通过storm.yaml配置是这样的:

  

drpc.servers:

- "drpc1.foo.com"

- "drpc2.foo.com"

最后,你通过StormSubmitter对象来提交DRPC topology(这个跟你提交其它topology没有区别)。如果要以远程的方式运行上面的例子,用下面的代码:

  

StormSubmitter.submitTopology("exclamation-drpc",  conf, builder.createRemoteTopology());

我们用createRemoteTopology方法来创建运行在真实集群上的DRPC topology。

更多精彩内容请关注:http://bbs.superwu.cn

关注超人学院微信二维码:

时间: 2024-10-07 17:54:01

Storm高级原语(二) -- DRPC详解的相关文章

iOS 开发之照片框架详解之二 —— PhotoKit 详解(下)

这里接着前文<iOS 开发之照片框架详解之二 —— PhotoKit 详解(上)>,主要是干货环节,列举了如何基于 PhotoKit 与 AlAssetLibrary 封装出通用的方法. 三. 常用方法的封装 虽然 PhotoKit 的功能强大很多,但基于兼容 iOS 8.0 以下版本的考虑,暂时可能仍无法抛弃 ALAssetLibrary,这时候一个比较好的方案是基于 ALAssetLibrary 和 PhotoKit 封装出一系列模拟系统 Asset 类的自定义类,然后在其中封装好兼容 A

iOS 开发之照片框架详解之二 —— PhotoKit 详解(上)

一. 概况 本文接着 iOS 开发之照片框架详解,侧重介绍在前文中简单介绍过的 PhotoKit 及其与 ALAssetLibrary 的差异,以及如何基于 PhotoKit 与 AlAssetLibrary 封装出通用的方法. 这里引用一下前文中对 PhotoKit 基本构成的介绍: PHAsset: 代表照片库中的一个资源,跟 ALAsset 类似,通过 PHAsset 可以获取和保存资源 PHFetchOptions: 获取资源时的参数,可以传 nil,即使用系统默认值 PHAssetCo

一步一步造个Ioc轮子(二),详解泛型工厂

一步一步造个Ioc轮子目录 .net core发布了,一步一步造个Ioc轮子,弄点.net魔法,近new的速度(一) 一步一步造个Ioc轮子(二),详解泛型工厂 详解泛型工厂 既然我说Ioc容器就是一个豪华版工厂,自动化装配的工厂,那我们就从工厂入手吧,先造个工厂,然后升级成Ioc容器 首先我们来写一个最最最简单的抽象工厂类,还是以前一篇的短信为例 public class SMSFactory { public static ISMS Get() { return new XSMS(); }

ViewPager 详解(二)---详解四大函数

前言:上篇中我们讲解了如何快速实现了一个滑动页面,但问题在于,PageAdapter必须要重写的四个函数,它们都各有什么意义,在上节的函数内部为什么要这么实现,下面我们就结合Android的API说明,详细讲解一下. 相关文章: 1.<ViewPager 详解(一)---基本入门> 2.<ViewPager 详解(二)---详解四大函数> 3.<ViewPager 详解(三)---PagerTabStrip与PagerTitleStrip添加标题栏的异同> 4.<

ant的高级使用,ant命令详解,ant打包,ant编译后打包去掉jar文件

在日常的项目开发中,我们可以经常性的需要打包测试,尤其是开发环境是windows,而实际环境则是linux. 这样的话,很多程序员要一会打一个包,一会打一个包,这些包可能会很大,实际上只有代码部分会变动,而jar包基本则不动. 当然很多人可能会说,自动化测试.很好的,我们今天要做的就是自动化测试的第一步. 这个时候我是可以使用ant来打包,去掉项目中的所有的jar文件.然后在项目的tomcat启动时到特定的地点去加载jar文件. 这样做,war包可能会只有几M,什么更小.具体如何动态的加载jar

DRPC详解

什么是DRPC? 分布式RPC(distributed RPC,DRPC)用于对Storm上大量的函数调用进行并行计算. 对于每一次函数调用,Storm集群上运行的拓扑接收调用函数的参数信息作为输入流,并将计算结果作为输出流发射出去. 一句话概括:Storm进行计算,根据客户端提交的请求参数,而返回Storm计算的结果. DRPC通过DRPC Server来实现,DRPC Server的整体工作过程如下: 接收到一个RPC调用请求: 发送请求到Storm上的拓扑: 从Storm上接收计算结果:

Struts2学习笔记二 配置详解

Struts2执行流程 1.简单执行流程,如下所示: 在浏览器输入请求地址,首先会被过滤器处理,然后查找主配置文件,然后根据地址栏中输入的/hello去每个package中查找为/hello的namespace,然后在包下寻找名为HelloAction的action,反射创建控制器HelloAction对象,调用method指定的方法hello,拿到返回值“success”,去result的配置节找success对应的页面(hello.js),呈现给用户. 2.Struts的架构,具体参考这里

Spring二 Bean详解

Bean详解 Spring框架的本质其实是:通过XML配置来驱动Java代码,这样就可以把原本由java代码管理的耦合关系,提取到XML配置文件中管理.这样就实现了系统中各组件的解耦,有利于后期的升级和维护.1.Bean的基本定义和Bean别名<beans>元素是Spring配置文件的根元素,该元素可以指定如下属性:default-lazy-init:指定<beans>元素下配置的所有bean默认的延迟初始化行为default-merge:指定<beans>元素下配置的

Struts2学习笔记(二)——配置详解

1.Struts2配置文件加载顺序: default.properties(默认常量配置) struts-default.xml(默认配置文件,主要配置bean和拦截器) struts-plugin.xml(配置插件) struts.xml(配置action或者常量等) struts.properties(常量配置) web.xml(配置JavaEE,如:监听器和过滤器) 2.Struts2配置文件详解 1)default.properties default.properties是Struts