Storm 流计算编程模型

曾经终败给现在 2022-08-19 08:56 179阅读 0赞

（一）基础介绍

随着互联网的更进一步发展，从Portal信息浏览型到Search信息搜索型到SNS关系交互传递型，以及电子商务、互联网旅游生活产品等将生活中的流通环节在线化。对效率的要求让大家对于实时性的要求进一步提升，而信息的交互和沟通正在从点对点往信息链甚至信息网的方向发展，这样必然带来数据在各个维度的交叉关联，数据爆炸已不可避免。因此流式处理加NoSQL产品应运而生，分别解决实时框架和数据大规模存储计算的问题。

当今世界，公司的日常运营经常会生成TB级别的数据。数据来源囊括了互联网装置可以捕获的任何类型数据，网站、社交媒体、交易型商业数据以及其它商业环境中创建的数据。考虑到数据的生成量，实时处理成为了许多机构需要面对的首要挑战。我们经常用的一个非常有效的开源实时计算工具就是Storm——Twitter开发，通常被比作“实时的Hadoop”。然而Storm远比Hadoop来的简单，因为用它处理大数据不会带来新老技术的交替。

Hadoop 在本质上是一个批处理系统。数据被引入 Hadoop 文件系统 (HDFS)并分发到各个节点进行处理。当处理完成时，结果数据返回到 HDFS供始发者使用。对比Hadoop的批处理，Storm是个实时的、分布式以及具备高容错的计算系统。同Hadoop一样Storm也可以处理大批量的数据，然而Storm在保证高可靠性的前提下还可以让处理进行的更加实时;也就是说，所有的信息都会被立即处理。Storm同样还具备容错和分布计算这些特性，这就让Storm可以扩展到不同的机器上进行大批量的数据处理。以下是Storm的几个特点：

*  编程模型简单

在大数据处理方面相信大家对hadoop已经耳熟能详，基于GoogleMap/Reduce来实现的Hadoop为开发者提供了map、reduce原语，使并行批处理程序变得非常地简单和优美。同样，Storm也为大数据的实时计算提供了一些简单优美的原语，这大大降低了开发并行实时处理的任务的复杂性，帮助你快速、高效的开发应用。

*  可扩展

在Storm集群中真正运行topology的主要有三个实体：工作进程、线程和任务。Storm集群中的每台机器上都可以运行多个工作进程，每个工作进程又可创建多个线程，每个线程可以执行多个任务，任务是真正进行数据处理的实体，我们开发的spout、bolt就是作为一个或者多个任务的方式执行的。

因此，计算任务在多个线程、进程和服务器之间并行进行，支持灵活的水平扩展。

*   高可靠性

Storm可以保证spout发出的每条消息都能被“完全处理”，这也是直接区别于其他实时系统的地方，如S4。

请注意，spout发出的消息后续可能会触发产生成千上万条消息，可以形象的理解为一棵消息树，其中spout发出的消息为树根，Storm会跟踪这棵消息树的处理情况，只有当这棵消息树中的所有消息都被处理了，Storm才会认为spout发出的这个消息已经被“完全处理”。如果这棵消息树中的任何一个消息处理失败了，或者整棵消息树在限定的时间内没有“完全处理”，那么spout发出的消息就会重发。

考虑到尽可能减少对内存的消耗，Storm并不会跟踪消息树中的每个消息，而是采用了一些特殊的策略，它把消息树当作一个整体来跟踪，对消息树中所有消息的唯一id进行异或计算，通过是否为零来判定spout发出的消息是否被“完全处理”，这极大的节约了内存和简化了判定逻辑，后面会对这种机制进行详细介绍。

这种模式，每发送一个消息，都会同步发送一个ack/fail，对于网络的带宽会有一定的消耗，如果对于可靠性要求不高，可通过使用不同的emit接口关闭该模式。

上面所说的，Storm保证了每个消息至少被处理一次，但是对于有些计算场合，会严格要求每个消息只被处理一次，幸而Storm的0.7.0引入了事务性拓扑，解决了这个问题，后面会有详述。

*   高容错性

如果在消息处理过程中出了一些异常，Storm会重新安排这个出问题的处理单元。Storm保证一个处理单元永远运行（除非你显式杀掉这个处理单元）。

当然，如果处理单元中存储了中间状态，那么当处理单元重新被Storm启动的时候，需要应用自己处理中间状态的恢复。

*   支持多种编程语言

除了用java实现spout和bolt，你还可以使用任何你熟悉的编程语言来完成这项工作，这一切得益于Storm所谓的多语言协议。多语言协议是Storm内部的一种特殊协议，允许spout或者bolt使用标准输入和标准输出来进行消息传递，传递的消息为单行文本或者是json编码的多行。

Storm支持多语言编程主要是通过ShellBolt,ShellSpout和ShellProcess这些类来实现的，这些类都实现了IBolt 和ISpout接口，以及让shell通过java的ProcessBuilder类来执行脚本或者程序的协议。

可以看到，采用这种方式，每个tuple在处理的时候都需要进行json的编解码，因此在吞吐量上会有较大影响。

*   支持本地模式

Storm有一种“本地模式”，也就是在进程中模拟一个Storm集群的所有功能，以本地模式运行topology跟在集群上运行topology类似，这对于我们开发和测试来说非常有用。

*   高效

用ZeroMQ作为底层消息队列,保证消息能快速被处理。目前已将内部消息通信机制更新为更快的Netty了， [http://yahooeng.tumblr.com/post/64758709722/making-storm-fly-with-netty][http_yahooeng.tumblr.com_post_64758709722_making-storm-fly-with-netty]

（二）概念简介

计算拓补： Topology

一个实时计算应用程序的逻辑在Storm里面被封装到topology对象里面,Storm的topology类似Hadoop的MR，但区别在于MR的JOB最终会终止，而topology除非显示的kill掉，否则用于不会终止。Topology其实就是一个Thrift结构，并且由Nimbus提供一个Thrift服务，可以使用任何语言向Storm创建并且提交topology。

消息流: Stream

Storm将消息流封装为一个有向无界的Tuple序列，tuple类型默认支持： integer,long, short, byte,  string, double, float,boolean和byte array。也可以实现序列化器，从而自定义类型。每个tuple被处理时都会生成一个id，使用OutputFieldsDeclarer申明Filed的Schema可以让Storm分配该tuple的默认id

消息源: Spout

Spouts是storm里面一个topology里面的消息生产者。Spout会从一个外部源读取数据并且向topology里面发射消息:tuple。消息源可以是可靠的也可以是不可靠的。一个可靠的消息源可以重新发射一个tuple，如果这个tuple没有被storm成功的处理，但是一个不可靠的消息源Spouts一旦发出一个tuple就把它彻底忘了。

nextTuple方法，spout最重要的方法！要么发射一个新的tuple到topology里面，或者简单的返回如果已经没有新的tuple了。由SpoutOutputCollector发射tuple

open方法，对象初始化时进行预置工作

close方法，对象销毁时进行清理工作，但Storm不保证该方法一定会被调用，因为，执行任务的机器可能随时宕机，或者任务被kill

deactivate方法，由外部用户触发bin/storm  deactivate  命令时调用。当用户执行deactivate时，对应Topology的spout会被deactivate，产生影响就是spout的nextTuple此后将不会被调用，直到用户再调用activate

activate方法，由外部用户触发bin/storm  activate命令时调用

ack方法，spout发送的所有tuple都会被acker跟踪，当acker监控到tuple被成功处理时，acker通知spout，该方法被调用

fail方法，spout发送的所有tuple都会被acker跟踪，当acker监控到tuple处理失败或超时的时候，acker通知spout，该方法被调用

declareOutputFields方法，用于申明spout向topology发送tuple的FiledSchema

getComponentConfiguratio n方法，用于修改当前组件（spout）的Config配置

消息处理者: Bolt

所有的消息处理逻辑被封装在bolts里面，例如连接、过滤、聚合、分组、查询数据库等。Bolts可以简单的做消息流的传递，也可以进行复杂的消息流处理，从而也就需要经过很多bolts。

execute方法,bolt最重要的方法！它以一个tuple作为输入进行处理，由OutputCollector的发射出处理后的tuple，以通知storm这个tuple被处理完成了。从而我们通知这个tuple的发射者Spouts。一般的流程是： Bolts处理一个输入tuple,   发射0个或者多个tuple,较为方便的方法是继承BaseRichBolt的实现。

prepare方法，用于在bolt处理tuple是进行预置工作

clean方法，设计为在本地模式LocalMode下执行，用于清理被打开的资源。但是集群不保证该方法被执行，因为，执行任务的机器可能随时宕机

declareOutputFields方法，用于定义bolt处理完tuple后，再向topology发送tuple时，申明tuple的FiledSchema。如果bolt要发射多条信息流，则使用declareStream定义stream，并在execute方法使用OutputCollector.emit来选择要发射的stream

getComponentConfiguratio n方法，用于修改当前组件（bolt）的Config配置

ISpoutOutputCollector与IOutputCollector

二者都是负责发送tuple的，但是由于一个是面向Spout，一个是面向Bolt的，它们的接口用途不同。

Spout通过调用ISpoutOutputCollector的emit函数进行tuple的发射，当然实际上emit函数并未完成实际的发送，它主要是根据用户提供的streamId，计算出该tuple需要发送到的目标taskID。emitDirect函数，更裸一些，直接指定目标taskID。它们都只是将组成的序列对放到一个队列中，然后会有另一个线程负责将tuple从队列中取出并发送到目标task。

IOutputCollector是会被Bolt调用的，与ISpoutOutputCollector功能类似。但是区别也很明显，首先我们可以看到它的emit系列函数，多了一个参数Collectionanchors，增加这样一个anchors原因在于，对于spout来说，它产生的tuple就是roottuple，但是对于bolt来说，它是通过一个或多个输入tuple，进而产生输出tuple的，这样tuple之间是有一个父子关系的，anchors就是用于指定当前要emit的这个tuple的所有父亲，正是通过它，才建立起tuple树，如果用户给了一个空的anchors，那么这个要emit的tuple将不会被加入tuple树，也就不会被追踪，即使后面它丢失了，也不会被spout感知。

除了anchors参数外，IOutputCollector的ack和fail的接口，与Spout的ack和fail完全不同，对于Spout来说ack和fail是提供给Spout在tuple发送成功或失败时进行处理的一个机会。而IOutputCollector的ack和fail则是向 acker汇报当前tuple的处理状态的，是需要Bolt在处理完tuple后主动调用的。

Stream Grouping： 消息分发策略

定义一个Topology的其中一步是定义每个bolt接受什么样的流作为输入。streamgrouping就是用来定义一个stream应该如果分配给Bolts上面的多个Tasks。Storm通过streamgrouping连接Spout和Bolt，从而形成一个有向的网状结构的Topology。

storm里面预定义有6种类型的stream grouping:

1.  Shuffle Grouping: 随机分组， 随机派发stream里面的tuple，保证每个bolt接收到的tuple数目相同。
2.  Fields Grouping：按字段分组， 比如按userid来分组，具有同样userid的tuple会被分到相同的Bolts， 而不同的userid则会被分配到不同的Bolts。
3.  All Grouping： 广播发送， 对于每一个tuple， 所有的Bolts都会收到。
4.  Global Grouping: 全局分组，这个tuple被分配到storm中的一个bolt的其中一个task。再具体一点就是分配给id值最低的那个task。
5.  Non Grouping: 不分组，这个分组的意思是说stream不关心到底谁会收到它的tuple。目前这种分组和Shuffle grouping是一样的效果，有一点不同的是storm会把这个bolt放到这个bolt的订阅者同一个线程里面去执行。
6.  Direct Grouping: 直接分组, 这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。只有被声明为DirectStream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的taskid(OutputCollector.emit方法也会返回taskid)

工作进程：Workers

Strom集群的每台服务器将会运行一个或多个工作进程Worker，每个工作进程Work执行整个topology的一个子集，并运行在自己的JVM中。每个工作进程Worker都隶属于一个特定的Topology，并为了一个或多个组件（spouts/bolts）而运行一个或多个executor。比如对于并行度是300的topology来说，如果我们使用50个工作进程来执行，那么每个工作进程会处理其中的6个tasks（每个工作进程分配6个线程）。storm会尽量均匀的工作分配给集群中所有的Supervisor的所有的Worker工作进程。

工作线程：Executor

Executor是由Woker产生的执行线程，运行于Worker的JVM中。每个Executor将会为相同的组件（spout/bolt）而运行一个或多个Task，每个Executor将使用 一个线程来运行它所有的Task，那么意味着这些 Task将在Executor连续执行。每个stream grouping则是定义怎么从一堆Executor  发射tuple到另外一堆Executor  。你可以调用TopologyBuilder.setSpout()和TopologyBuilder.setBolt来设置并行度 — 也就是控制整个Topology中有多少个Executor  。

任务：Tasks

实际执行逻辑的运行单元，每个组件（spout/bolt）将运行很多task来完成任务。每个组件（spout/bolt）的任务数Task是固定设置的，但执行线程数Executor却是可变的，即Topology运行期间Task是静态不变的，而Executor线程数是可动态分配的。所以一般情况下，从数量上 threads <=tasks，也就是说Executor线程数越多，由其产生的且运行在其Executor线程中执行的Task数越少，进一步也就等价于Task的并发执行度越高，资源占用率越大，但Task的执行效率和延迟越低！

在Storm默认配置下，task将会与executor数量保持一致，即每个executor运行一个task。如下代码：

topologyBuilder.setSpout("blue-spout", new BlueSpout(), 2); // parallelism hint

topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2) .setNumTasks(4) .shuffleGrouping("blue-spout);

topologyBuilder.setBolt("yellow-bolt", new YellowBolt(), 6) .shuffleGrouping("green-bolt");

这里blue-sport设置了2个executor线程，则默认每个线程执行一个task；而green-bolt设置了2个executor，4个task；最后yellow-bolt设置了6个executor线程，默认每个线程执行一个task。如果此时Storm集群存在2个Worker，那么共计10个Executor将在每个Worker上面分配5个Executor线程，其中包括1个blue-spout，1个green-blout和3个yellow-bolt，而每个green-blout的Executor上面运行了2个task。

并行计算配置顺序

defaults.yaml < storm.yaml < topology-specificconfiguration < internal component-specific configuration <external component-specificconfiguration。通常Config.TOPOLOGY\_MAX\_TASK\_PARALLELISM在测试时进行设置，用于设置每个组件（spout/bolt）派生的最大线程数Executor

运行期间调整并发设置

1、通过Storm UI     2、通过Storm CLI

# Reconfigure the topology "mytopology" to use 5 worker processes,
     # the spout "blue-spout" to use 3 executors and
     # the bolt "yellow-bolt" to use 10 executors.
    
     $ storm rebalance mytopology -n 5 -e blue-spout=3 -e yellow-bolt=10

Storm集群

控制节点(Nimbus)和工作节点(Supervisor)。Nimbus是主控节点监控后台进程，作用类似Hadoop里面的JobTracker。Nimbus负责在集群里面分布代码，分配工作给机器，并且监控集群状态。

Supervisor是工作节点监控后台进程，作用是监听分配给它那台机器的任务，根据需要启动/关闭工作进程Worker。每一个工作进程Worker执行一个Topology的一个子集；一个运行的Topology由运行在很多Supervisor机器上的很多工作进程Worker组成。

高可用性容错机制

Task：Spout、Bolt挂掉或超时，Nimbus将会把Task分配到其他主机运行

Worker：如果Worker挂掉，Supervisor将尝试对他重启，如果持续出错，超过超时时间与Nimbus连接，Nimbus将会把Worker分配到其他主机运行

Nimbus与Supervisor：其设计与Zookeeper一样，都是fast-fail的系统架构，它们如果挂掉，那么重启后就像没事儿一样。特别注意，这与Hadoop不同，一个JobStracker挂掉，上面运行的所有Job将会丢失。

Nimbus单点：失去Nimbus，Worker依然可以正常工作。另外，如果Worker挂掉，Supervisor依然可以尝试在本机重启Worker。然而，没有Nimbus，Worker无法分配到其他主机重启。Nimbus和Supervisor都是后台守护进程，在实际生产环境中，失去守护进程不会产生太大灾难。

高可靠消息

Storm可以保证Tuple以及被这个Tuple派生出来的所有Tuple被完全成功处理，每个Tuple都有一个msgId，在task派生Tuple时，通过指定父级Tuple完成锚定（anchoring），子Tuple处理完成通过ack方法通知Storm，子Tuple处理失败或超时通过fail方法通知Storm，Storm除了spout和bolt两类task以外，还有一类特殊的task-acker进行跟踪Tuple树，Storm通过一致性哈希算法将Tuple的msgId分配给acker进行跟踪，acker保存了spout和bolt的task-id以及64位长度tuple-msgId的映射关系，通过异或算法保证20bytes内存即可跟踪一颗Tuple树，当异或值为0时，即表示该Tuple树已完全处理成功。

collector.emit(new Values(word));--脱离Tuple树，无高可靠保证，即被处理完成后，即被忘记

collector.emit(tuple, newValues(word));  \--锚定Tuple树，有高可靠保证，即处理发生失败，被锚定的父级Tuple将重新处理所有派生的Tuple

一般有以下三种失败场景：

1、spout挂掉，由发送该消息的消息源重新发送该消息

2、bolt挂掉，在超时之后，标记为失败，等待重新处理

3、acker挂掉，由这个acker跟踪的整棵Tuple树标记为失败，等待重新处理，ackertask是非常轻量级的，可以通过Strom UI监控和调整acker数量。

关闭高可靠消息一般有三种方式：

1、全局设置不跟踪Tuple树，即把Config.TOPOLOGY\_ACKERS 设置成0。在这种情况下，storm会在spout发射一个tuple之后马上调用spout的ack方法。也就是说这个tuple树不会被跟踪。

2、Spout Tuple层面去掉可靠性。即可以在发射tuple的时候不指定msgId来达到不跟粽某个特定的spout tuple的目的。

3、BoltTuple层面去掉可靠性，即对一颗tuple树里面的某一部分到底成不成功不是很关心，那么可以在发射这些tuple的时候unanchor它们。这样这些tuple就不在tuple树里面， 也就不会被跟踪了。

批处理CoordinatedBolt

Storm最上层的跟节点Tuple通过ack判断其是否处理完了所有的tuple，而其下游所有派生出来的子Tuple都通过其父类通知。这个通知机制由CoordinatedBolt所封装，一个bolt在接到所有的上游task发送的tuple个数信息之后，对比它接收到的tuple数量，如果数量对上了，说明它接收到了所有的tuple— 它处理完成了。这样它处理完成了，它可以重复上面的步骤通知它的下游，它的下游再通知它的下游的下游等等。

总结一下，每个tuple怎么知道自己处理完成了的？都是靠它的上游通知的。所以只要一个bolt有上游，它就能够知道自己什么时候完成。

这常常可以用于批处理、分布式DRPC、事务处理 [https://github.com/nathanmarz/storm/wiki/Transactional-topologies][https_github.com_nathanmarz_storm_wiki_Transactional-topologies]等

（三）Storm常见模式

[http://www.cnblogs.com/panfeng412/tag/Storm/][http_www.cnblogs.com_panfeng412_tag_Storm]

（四）DRPC

Storm里面引入DRPC主要是利用storm的实时计算能力来并行化CPUintensive的计算。DRPC的stormtopology以Function参数流作为输入，而把这些Function函数调用的返回值作为topology的输出流。

DRPC其实不能算是storm本身的一个特性，它是通过组合storm的原语spout，bolt，topology而成的一种模式(pattern)。本来应该把DRPC单独打成一个包的，但是DRPC实在是太有用了，所以把它和storm捆绑在一起。一个DRPC客户端代码实现如下：

DRPCClient client = new DRPCClient("drpc-host", 3772);
    String result = client.execute("reach", "http://twitter.com");

1）DRPC工作流程

Distributed RPC的工作流大致如下：

*   接收一个RPC请求
 *   发送请求到stormtopology 
 *   从storm topology接收结果
 *   把结果发回给等待的客户端

因此，从客户端的角度来看一个DRPC调用跟一个普通的RPC调用没有任何区别。

![Storm <wbr>流计算编程模型][Storm_wbr]

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

这里以前面Storm Starter的对word单词增加“！”为例，Bolt实现代码如下：

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"));
        }
    }

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

*  设置spout--（DRPCSpout）
 *  把结果返回给DRPC服务器 
 *  给bolt提供有限聚合几组tuples的能力

public static void main(String[] args) throws Exception { 
        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("exclamation");
        builder.addBolt(new ExclaimBolt(), 3);
        // ...
    }

可以看出来，我们需要做的事情非常的少。创建LinearDRPCTopologyBuilde r的时候，你需要申明实现DRPC调用函数的名字（一个DRPC服务器可以协调很多函数，函数与函数之间靠函数名字来区分），该名字将会被客户端调用时使用。这里声明addBolt的第一个bolt会接收两维tuple，tuple的第一个field是request-id，第二个field是这个请求的实际参数。LinearDRPCTopologyBuilde r同时要求我们在topology的最后一个bolt发射一个二维tuple：第一个field是request-id，第二个field是这个请求调用返回的结果。最后所有中间过程tuple的第一个field必须是request-id。

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

2）本地模式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服务器，在本地模式里面LocalDRPC对象不与任何端口绑定，所以我们的topology对象需要知道和谁交互。这就是为什么createLocalTopology方法接受一个LocalDRPC对象作为输入的原因。

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

3）远程模式DRPC

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

*   启动DRPC服务器 bin/storm drpc
 *  配置DRPC服务器的地址

DRPCSpout需要这个地址从而可以从DRPC服务器来接收函数调用，通过storm.yaml进行如下配置   
drpc.servers:   
   \- "drpc1.foo.com"   
   \- "drpc2.foo.com"

*  提交DRPC topology到storm集群里面去

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

4）DRPC架构原理

1.  首先，DRPCClient提交请求到DRPCServer 
2.  其次，DRPCServer首先给这个请求产生一个request-id, 然后把它丢到一个request-id -> request池子里面。在DRPCServer在把request放入池子里面的时候，会同时生成一个Semaphore,并且把这个Semaphore把放到一个 request-id ->semaphore池子里面；同时它调用semaphore.acquire()来等在这个semaphore上面等待结果的到来。 
3.  再次，Storm组件从 request-id -> request池子中获取需要处理的请求，通过DRPCSpout, PreapreRequest, JoinResult,ReturnResults等一帮家伙去处理这个请求，然后把处理完的请求结果发回到DRPCServer的 request-id-> result池子里面去，最后会通过request-id去 request-id ->semaphore池子里面取出这个请求所对应的semaphore,并且调用semaphore.release()来释放这个semaphore 
4.  最后，在semaphore得到释放之后，DRPCServer上面阻塞的等待线程得以继续执行，去request-id -> result池子里面把结果取出来，返回给等待的客户端。

5）LinearDRPCTopologyBuilde r的工作原理

> 1.DRPCSpout发射tuple: \[args, return-info\]。return-info包含DRPC服务器的主机地址、端口以及当前请求的request-id 
> 
> 2.DRPC Topology包含以下元素：

*  DRPCSpout 
 *  PrepareRequest(生成request-id, returninfo以及args) 
 *  CoordinatedBolt 
 *  JoinResult -- 组合结果和return info 
 *  ReturnResult -- 连接到DRPC服务器并且返回结果

> 3.LinearDRPCTopologyBuilde r是利用storm的原语来构建高层抽象的很好的例子。
> 
>   
>

6）非线性DRPC Topology

LinearDRPCTopologyBuilde r只能搞定"线性"的DRPCtopology。所谓的线性就是说你的计算过程是一步接着一步， 串联。我们不难想象还有其它的可能 --并联（回想一下初中物理里面学的并联电路吧), 现在你如果想解决这种这种并联的case的话，那么你需要自己去使用CoordinatedBolt来处理所有的事情了。所幸的是Trident已将其功能纳入，因此可以直接结合Trident使用DRPC。

7）异步DRPC

目前Storm现在还不支持异步的DRPC,因此向DPRC发起请求后会一直同步等待响应。如果请求执行耗时，建议设置一个超时时间，以免整个应用挂起！不过作者阐述在目前的设计模型的基础上去实现异步的DRPC应该是很简单的，相信在Storm0.8.2随后的版本中很快就会得到实现。

[http_yahooeng.tumblr.com_post_64758709722_making-storm-fly-with-netty]: http://yahooeng.tumblr.com/post/64758709722/making-storm-fly-with-netty
[https_github.com_nathanmarz_storm_wiki_Transactional-topologies]: https://github.com/nathanmarz/storm/wiki/Transactional-topologies
[http_www.cnblogs.com_panfeng412_tag_Storm]: http://www.cnblogs.com/panfeng412/tag/Storm/
[Storm_wbr]: