问题导读：

1.hadoop有master与slave，Storm与之对应的节点是什么？

2.Storm控制节点上面运行一个后台程序被称之为什么？

3.Supervisor的作用是什么？

4.Topology与Worker之间的关系是什么？

5.Nimbus和Supervisor之间的所有协调工作有master来完成，还是Zookeeper集群完成？

6.storm稳定的原因是什么？

7.如何运行Topology？

strom jar all-your-code.jar backtype.storm.MyTopology arg1 arg2

8.spout是什么？

9.bolt是什么？

10.Topology由两部分组成？

11.stream grouping有几种？

伴随着信息科技日新月异的发展，信息呈现出爆发式的膨胀，人们获取信息的途径也更加多样、更加便捷，同时对于信息的时效性要求也越来越高。举个搜索场景中的例子，当一个卖家发布了一条宝贝信息时，他希望的当然是这个宝贝马上就可以被卖家搜索出来、点击、购买啦，相反，如果这个宝贝要等到第二天或者更久才可以被搜出来，估计这个大哥就要不满意了。再举一个推荐的例子，如果用户昨天在淘宝上买了一双袜子，今天想买一副泳镜去游泳，但是却发现系统在不遗余力地给他推荐袜子、鞋子，根本对他今天寻找泳镜的行为视而不见，估计这哥们心里就会不痛快了呀。其实稍微了解点背景知识的码农们都知道，这是因为后台系统做的是每天一次的全量处理，而且大多是在夜深人静之时做的，那么你今天白天做的事情当然要明天才能反映出来啦。

一、一个实时计算系统要解决的问题

全量数据处理使用的大多是鼎鼎大名的hadoop或者hive，作为一个批处理系统，hadoop以其吞吐量大、自动容错等优点，在海量数据处理上得到了广泛的使用。但是，hadoop不擅长实时计算，因为它天然就是为批处理而生的，这也是业界一致的共识。否则最近这两年也不会有s4,storm,puma这些实时计算系统如雨后春笋般冒出来啦。先抛开s4,storm,puma这些系统不谈，我们首先来看一下，如果让我们自己设计一个实时计算系统，我们要解决哪些问题。

1. 低延迟。都说了是实时计算系统了，延迟是一定要低的。
2. 高性能。性能不高就是浪费机器，浪费机器是要受批评的哦。
3. 分布式。系统都是为应用场景而生的，如果你的应用场景、你的数据和计算单机就能搞定，那么不用考虑这些复杂的问题了。我们所说的是单机搞不定的情况。
4. 可扩展。伴随着业务的发展，我们的数据量、计算量可能会越来越大，所以希望这个系统是可扩展的。
5. 容错。这是分布式系统中通用问题。一个节点挂了不能影响我的应用。

好，如果仅仅需要解决这5个问题，可能会有无数种方案，而且各有千秋，随便举一种方案，使用消息队列+分布在各个机器上的工作进程就ok啦。我们再继续往下看。

1. 容易在上面开发应用程序。亲，你设计的系统需要应用程序开发人员考虑各个处理组件的分布、消息的传递吗？如果是，那有点麻烦啊，开发人员可能会用不好，也不会想去用。
2. 消息不丢失。用户发布的一个宝贝消息不能在实时处理的时候给丢了，对吧？更严格一点，如果是一个精确数据统计的应用，那么它处理的消息要不多不少才行。这个要求有点高哦。

在2011年Storm开源之前，由于Hadoop的火红，整个业界都在喋喋不休地谈论大数据。Hadoop的高吞吐，海量数据处理的能力使得人们可以方便地处理海量数据。但是，Hadoop的缺点也和它的优点同样鲜明——延迟大，响应缓慢，运维复杂。

有需求也就有创造，在Hadoop基本奠定了大数据霸主地位的时候，很多的开源项目都是以弥补Hadoop的实时性为目标而被创造出来。而在这个节骨眼上Storm横空出世了。

Storm带着流式计算的标签华丽丽滴出场了，看看它的一些卖点：

• 分布式系统：可横向拓展,现在的项目不带个分布式特性都不好意思开源。
• 运维简单：Storm的部署的确简单。虽然没有Mongodb的解压即用那么简单，但是它也就是多安装两个依赖库而已。
• 高度容错：模块都是无状态的，随时宕机重启。
• 无数据丢失：Storm创新性提出的ack消息追踪框架和复杂的事务性处理,能够满足很多级别的数据处理需求。不过，越高的数据处理需求，性能下降越严重。
• 多语言：实际上，Storm的多语言更像是临时添加上去似的。因为，你的提交部分还是要使用Java实现。

二、简单认识Storm

Storm是一个免费开源、分布式、高容错的实时计算系统。Storm令持续不断的流计算变得容易，弥补了Hadoop批处理所不能满足的实时要求。Storm经常用于在实时分析、在线机器学习、持续计算、分布式远程调用和ETL等领域。Storm的部署管理非常简单，而且，在同类的流式计算工具，Storm的性能也是非常出众的。Storm对于实时计算的的意义相当于Hadoop对于批处理的意义。Hadoop为我们提供了Map和Reduce原语，使我们对数据进行批处理变的非常的简单和优美。同样，Storm也对数据的实时计算提供了简单Spout和Bolt原语。

Storm适用的场景：

1、流数据处理：Storm可以用来用来处理源源不断的消息，并将处理之后的结果保存到持久化介质中。

2、分布式RPC：由于Storm的处理组件都是分布式的，而且处理延迟都极低，所以可以Storm可以做为一个通用的分布式RPC框架来使用。

在这个教程里面我们将学习如何创建Topologies, 并且把topologies部署到storm的集群里面去。Java将是我们主要的示范语言， 个别例子会使用python以演示storm的多语言特性。

1、一个Storm集群的基本组件

storm的集群表面上看和hadoop的集群非常像。但是在Hadoop上面你运行的是MapReduce的Job, 而在Storm上面你运行的是Topology。它们是非常不一样的 — 一个关键的区别是： 一个MapReduce Job最终会结束， 而一个Topology永远运行（除非你显式的杀掉他）。

在Storm的集群里面有两种节点： 控制节点(master node)和工作节点(worker node)。

控制节点上面运行一个后台程序：Nimbus， 它的作用类似Hadoop里面的JobTracker。Nimbus负责在集群里面分布代码，分配工作给机器， 并且监控状态。

每一个工作节点上面运行一个叫做Supervisor的节点（类似 TaskTracker）。Supervisor会监听分配给它那台机器的工作，根据需要 启动/关闭工作进程。每一个工作进程执行一个Topology（类似 Job）的一个子集；一个运行的Topology由运行在很多机器上的很多工作进程 Worker（类似 Child）组成。

                                                  storm topology结构

                          Storm VS MapReduce

Nimbus和Supervisor之间的所有协调工作都是通过一个Zookeeper集群来完成。并且，nimbus进程和supervisor都是快速失败（fail-fast)和无状态的。所有的状态要么在Zookeeper里面， 要么在本地磁盘上。这也就意味着你可以用kill -9来杀死nimbus和supervisor进程， 然后再重启它们，它们可以继续工作， 就好像什么都没有发生过似的。这个设计使得storm不可思议的稳定。

• Nimbus负责在集群里面发送代码，分配工作给机器，并且监控状态。全局只有一个。
• Supervisor会监听分配给它那台机器的工作，根据需要启动/关闭工作进程Worker。每一个要运行Storm的机器上都要部署一个，并且按照机器的配置设定上面分配的槽位数。
• Zookeeper是Storm重点依赖的外部资源。Nimbus和Supervisor甚至实际运行的Worker都是把心跳保存在Zookeeper上的。Nimbus也是根据Zookeerper上的心跳和任务运行状况，进行调度和任务分配的。
• Storm提交运行的程序称为Topology。
• Topology处理的最小的消息单位是一个Tuple，也就是一个任意对象的数组。
• Topology由Spout和Bolt构成。Spout是发出Tuple的结点。Bolt可以随意订阅某个Spout或者Bolt发出的Tuple。Spout和Bolt都统称为component。

2、Topologies

为了在storm上面做实时计算， 你要去建立一些topologies。一个topology就是一个计算节点所组成的图。Topology里面的每个处理节点都包含处理逻辑， 而节点之间的连接则表示数据流动的方向。

下图是一个Topology设计的逻辑图的例子。

3、Stream

Stream是storm里面的关键抽象。一个stream是一个没有边界的tuple序列。storm提供一些原语来分布式地、可靠地把一个stream传输进一个新的stream。比如： 你可以把一个tweets流传输到热门话题的流。

storm提供的最基本的处理stream的原语是spout和bolt。你可以实现Spout和Bolt对应的接口以处理你的应用的逻辑。

spout的流的源头。比如一个spout可能从Kestrel队列里面读取消息并且把这些消息发射成一个流。又比如一个spout可以调用twitter的一个api并且把返回的tweets发射成一个流。

通常Spout会从外部数据源（队列、数据库等）读取数据，然后封装成Tuple形式，之后发送到Stream中。Spout是一个主动的角色，在接口内部有个nextTuple函数，Storm框架会不停的调用该函数。

bolt可以接收任意多个输入stream， 作一些处理， 有些bolt可能还会发射一些新的stream。一些复杂的流转换， 比如从一些tweet里面计算出热门话题， 需要多个步骤， 从而也就需要多个bolt。 Bolt可以做任何事情: 运行函数， 过滤tuple, 做一些聚合， 做一些合并以及访问数据库等等。

Bolt处理输入的Stream，并产生新的输出Stream。Bolt可以执行过滤、函数操作、Join、操作数据库等任何操作。Bolt是一个被动的角色，其接口中有一个execute(Tuple input)方法，在接收到消息之后会调用此函数，用户可以在此方法中执行自己的处理逻辑。

spout和bolt所组成一个网络会被打包成topology， topology是storm里面最高一级的抽象（类似 Job）， 你可以把topology提交给storm的集群来运行。topology的结构在Topology那一段已经说过了，这里就不再赘述了。

topology结构

topology里面的每一个节点都是并行运行的。 在你的topology里面， 你可以指定每个节点的并行度， storm则会在集群里面分配那么多线程来同时计算。

一个topology会一直运行直到你显式停止它。storm自动重新分配一些运行失败的任务， 并且storm保证你不会有数据丢失， 即使在一些机器意外停机并且消息被丢掉的情况下。

4、数据模型(Data Model)

storm使用tuple来作为它的数据模型。每个tuple是一堆值，每个值有一个名字，并且每个值可以是任何类型， 在我的理解里面一个tuple可以看作一个没有方法的java对象。总体来看，storm支持所有的基本类型、字符串以及字节数组作为tuple的值类型。你也可以使用你自己定义的类型来作为值类型， 只要你实现对应的序列化器(serializer)。

一个Tuple代表数据流中的一个基本的处理单元，例如一条cookie日志，它可以包含多个Field，每个Field表示一个属性。

Tuple本来应该是一个Key-Value的Map，由于各个组件间传递的tuple的字段名称已经事先定义好了，所以Tuple只需要按序填入各个Value，所以就是一个Value List。

一个没有边界的、源源不断的、连续的Tuple序列就组成了Stream。

topology里面的每个节点必须定义它要发射的tuple的每个字段。 比如下面这个bolt定义它所发射的tuple包含两个字段，类型分别是: double和triple。

public class DoubleAndTripleBolt implements IRichBolt {    private OutputCollectorBase _collector;     @Override    public void prepare(Map conf, TopologyContext context, OutputCollectorBase collector) {        _collector = collector;    }     @Override    public void execute(Tuple input) {        int val = input.getInteger(0);        _collector.emit(input,newValues(val*2, val*3));        _collector.ack(input);    }     @Override    public void cleanup() {    }     @Override    public void declare OutputFields(OutputFieldsDeclarer declarer) {        declarer.declare(newFields("double","triple"));    }}

declareOutputFields方法定义要输出的字段 ： ["double", "triple"]。这个bolt的其它部分我们接下来会解释。

5、一个简单的Topology

让我们来看一个简单的topology的例子， 我们看一下storm-starter里面的ExclamationTopology:

TopologyBuilder builder =new TopologyBuilder();builder.setSpout(1,new TestWordSpout(),10);builder.setBolt(2,new ExclamationBolt(),3)        .shuffleGrouping(1);builder.setBolt(3,new ExclamationBolt(),2)        .shuffleGrouping(2);

这个Topology包含一个Spout和两个Bolt。Spout发射单词， 每个bolt在每个单词后面加个”!!!”。这三个节点被排成一条线: spout发射单词给第一个bolt， 第一个bolt然后把处理好的单词发射给第二个bolt。如果spout发射的单词是["bob"]和["john"], 那么第二个bolt会发射["bolt!!!!!!"]和["john!!!!!!"]出来。

我们使用setSpout和setBolt来定义Topology里面的节点。这些方法接收我们指定的一个id， 一个包含处理逻辑的对象(spout或者bolt), 以及你所需要的并行度。

这个包含处理的对象如果是spout那么要实现IRichSpout的接口， 如果是bolt，那么就要实现IRichBolt接口.

最后一个指定并行度的参数是可选的。它表示集群里面需要多少个thread来一起执行这个节点。如果你忽略它那么storm会分配一个线程来执行这个节点。

setBolt方法返回一个InputDeclarer对象， 这个对象是用来定义Bolt的输入。 这里第一个Bolt声明它要读取spout所发射的所有的tuple — 使用shuffle grouping。而第二个bolt声明它读取第一个bolt所发射的tuple。shuffle grouping表示所有的tuple会被随机的分发给bolt的所有task。给task分发tuple的策略有很多种，后面会介绍。

如果你想第二个bolt读取spout和第一个bolt所发射的所有的tuple， 那么你应该这样定义第二个bolt:

builder.setBolt(3,new ExclamationBolt(),5)            .shuffleGrouping(1)            .shuffleGrouping(2);

让我们深入地看一下这个topology里面的spout和bolt是怎么实现的。Spout负责发射新的tuple到这个topology里面来。TestWordSpout从["nathan", "mike", "jackson", "golda", "bertels"]里面随机选择一个单词发射出来。TestWordSpout里面的nextTuple()方法是这样定义的：

public void nextTuple() {    Utils.sleep(100);    final String[] words =new String[] {"nathan","mike",                     "jackson","golda","bertels"};    final Random rand =new Random();    final String word = words[rand.nextInt(words.length)];    _collector.emit(new Values(word));}

可以看到，实现很简单。

ExclamationBolt把”!!!”拼接到输入tuple后面。我们来看下ExclamationBolt的完整实现。

public static class ExclamationBolt implements IRichBolt {    OutputCollector _collector;     public void prepare(Map conf, TopologyContext context,                        OutputCollector collector) {        _collector = collector;    }     public void execute(Tuple tuple) {        _collector.emit(tuple,new Values(tuple.getString(0) +"!!!"));        _collector.ack(tuple);    }     public void cleanup() {    }     public void declare OutputFields(OutputFieldsDeclarer declarer) {        declarer.declare(new Fields("word"));    }}

prepare方法提供给bolt一个Outputcollector用来发射tuple。Bolt可以在任何时候发射tuple — 在prepare, execute或者cleanup方法里面, 或者甚至在另一个线程里面异步发射。这里prepare方法只是简单地把OutputCollector作为一个类字段保存下来给后面execute方法使用。

execute方法从bolt的一个输入接收tuple(一个bolt可能有多个输入源). ExclamationBolt获取tuple的第一个字段，加上”!!!”之后再发射出去。如果一个bolt有多个输入源，你可以通过调用Tuple#getSourceComponent方法来知道它是来自哪个输入源的。

execute方法里面还有其它一些事情值得一提： 输入tuple被作为emit方法的第一个参数，并且输入tuple在最后一行被ack。这些呢都是Storm可靠性API的一部分，后面会解释。

cleanup方法在bolt被关闭的时候调用， 它应该清理所有被打开的资源。但是集群不保证这个方法一定会被执行。比如执行task的机器down掉了，那么根本就没有办法来调用那个方法。cleanup设计的时候是被用来在local mode的时候才被调用(也就是说在一个进程里面模拟整个storm集群), 并且你想在关闭一些topology的时候避免资源泄漏。

最后，declareOutputFields定义一个叫做”word”的字段的tuple。

以local mode运行ExclamationTopology

让我们看看怎么以local mode运行ExclamationToplogy。

storm的运行有两种模式: 本地模式和分布式模式. 在本地模式中， storm用一个进程里面的线程来模拟所有的spout和bolt. 本地模式对开发和测试来说比较有用。 你运行storm-starter里面的topology的时候它们就是以本地模式运行的， 你可以看到topology里面的每一个组件在发射什么消息。

在分布式模式下， storm由一堆机器组成。当你提交topology给master的时候， 你同时也把topology的代码提交了。master负责分发你的代码并且负责给你的topolgoy分配工作进程。如果一个工作进程挂掉了， master节点会把它重新分配到其它节点。关于如何在一个集群上面运行topology， 你可以看看Running topologies on a production cluster文章。

下面是以本地模式运行ExclamationTopology的代码:

Config conf =new Config();conf.setDebug(true);conf.setNumWorkers(2); LocalCluster cluster =new LocalCluster();cluster.submitTopology("test", conf, builder.createTopology());Utils.sleep(10000);cluster.killTopology("test");cluster.shutdown();

首先， 这个代码定义通过定义一个LocalCluster对象来定义一个进程内的集群。提交topology给这个虚拟的集群和提交topology给分布式集群是一样的。通过调用submitTopology方法来提交topology， 它接受三个参数：要运行的topology的名字，一个配置对象以及要运行的topology本身。

topology的名字是用来唯一区别一个topology的，这样你然后可以用这个名字来杀死这个topology的。前面已经说过了， 你必须显式的杀掉一个topology， 否则它会一直运行。

Conf对象可以配置很多东西， 下面两个是最常见的：

• TOPOLOGY_WORKERS(setNumWorkers) 定义你希望集群分配多少个工作进程给你来执行这个topology. topology里面的每个组件会被需要线程来执行。每个组件到底用多少个线程是通过setBolt和setSpout来指定的。这些线程都运行在工作进程里面. 每一个工作进程包含一些节点的一些工作线程。比如， 如果你指定300个线程，60个进程， 那么每个工作进程里面要执行6个线程， 而这6个线程可能属于不同的组件(Spout, Bolt)。你可以通过调整每个组件的并行度以及这些线程所在的进程数量来调整topology的性能。
• TOPOLOGY_DEBUG(setDebug), 当它被设置成true的话， storm会记录下每个组件所发射的每条消息。这在本地环境调试topology很有用， 但是在线上这么做的话会影响性能的。

运行中的Topology主要由以下三个组件组成的：

Worker processes（工作进程）——Spout/Bolt中运行具体处理逻辑的进程

Executors (threads)（线程、执行器）——物理线程

Tasks（任务）——具体的处理逻辑对象

下图简要描述了这3者之间的关系：

　　storm集群的一个节点可能有一个或者多个工作进程（worker）运行在一个多个拓扑上，一个工作进程执行拓扑的一个子集。工作进程（worker）属于一个特定的拓扑，并可能为这个拓扑的一个或者多个组件（spout/bolt）运行一个或多个执行器（executor线程）。一个运行中的拓扑包括多个运行在storm集群内多个节点的进程。

　　1个worker进程执行的是1个topology的子集（注：不会出现1个worker为多个topology服务）。1个worker进程会启动1个或多个executor线程来执行1个topology的component(spout或bolt)。因此，1个运行中的topology就是由集群中多台物理机上的多个worker进程组成的。

　　executor是1个被worker进程启动的单独线程。每个executor只会运行1个topology的1个component(spout或bolt)的task（注：task可以是1个或多个，storm默认是1个component只生成1个task，executor线程里会在每次循环里顺序调用所有task实例）。

　　task是最终运行spout或bolt中代码的单元（注：1个task即为spout或bolt的1个实例，executor线程在执行期间会调用该task的nextTuple或execute方法）。topology启动后，1个component(spout或bolt)的task数目是固定不变的，但该component使用的executor线程数可以动态调整（例如：1个executor线程可以执行该component的1个或多个task实例）。这意味着，对于1个component存在这样的条件：#threads<=#tasks（即：线程数小于等于task数目）。默认情况下task的数目等于executor线程数目，即1个executor线程只运行1个task。

Spout或者Bolt的Task个数一旦指定之后就不能改变了，而Executor的数量可以根据情况来进行动态的调整。默认情况下# executor = #tasks即一个Executor中运行着一个Task

6、流分组策略(Stream grouping)

流分组策略告诉topology如何在两个组件之间发送tuple。 要记住， spouts和bolts以很多task的形式在topology里面同步执行。如果从task的粒度来看一个运行的topology， 它应该是这样的:

从task角度来看topology

当Bolt A的一个task要发送一个tuple给Bolt B， 它应该发送给Bolt B的哪个task呢？

stream grouping专门回答这种问题的。在我们深入研究不同的stream grouping之前， 让我们看一下里面的另外一个topology。WordCountTopology读取一些句子， 输出句子里面每个单词出现的次数.

TopologyBuilder builder =new TopologyBuilder(); builder.setSpout(1,new RandomSentenceSpout(),5);builder.setBolt(2,new SplitSentence(),8)        .shuffleGrouping(1);builder.setBolt(3,new WordCount(),12)        .fieldsGrouping(2,new Fields("word"));

SplitSentence对于句子里面的每个单词发射一个新的tuple, WordCount在内存里面维护一个单词->次数的mapping， WordCount每收到一个单词， 它就更新内存里面的统计状态。

有好几种不同的stream grouping:

• 最简单的grouping是shuffle grouping, 它随机发给任何一个task。上面例子里面RandomSentenceSpout和SplitSentence之间用的就是shuffle grouping, shuffle grouping对各个task的tuple分配的比较均匀。
• 一种更有趣的grouping是fields grouping, SplitSentence和WordCount之间使用的就是fields grouping, 这种grouping机制保证相同field值的tuple会去同一个task， 这对于WordCount来说非常关键，如果同一个单词不去同一个task， 那么统计出来的单词次数就不对了。

fields grouping是stream合并，stream聚合以及很多其它场景的基础。在背后呢， fields grouping使用的一致性哈希来分配tuple的。

还有一些其它类型的stream grouping. 你可以在Concepts一章里更详细的了解。

下面是一些常用的 “路由选择” 机制：

Storm的Grouping即消息的Partition机制。当一个Tuple被发送时，如何确定将它发送个某个（些）Task来处理？？

l ShuffleGrouping：随机选择一个Task来发送。

l FiledGrouping：根据Tuple中Fields来做一致性hash，相同hash值的Tuple被发送到相同的Task。

l AllGrouping：广播发送，将每一个Tuple发送到所有的Task。

l GlobalGrouping：所有的Tuple会被发送到某个Bolt中的id最小的那个Task。

l NoneGrouping：不关心Tuple发送给哪个Task来处理，等价于ShuffleGrouping。

l DirectGrouping：直接将Tuple发送到指定的Task来处理。, 这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。 只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过 TopologyContext来获取处理它的消息的taskid (OutputCollector.emit方法也会返回taskid)。

7、使用别的语言来定义Bolt

Bolt可以使用任何语言来定义。用其它语言定义的bolt会被当作子进程(subprocess)来执行， storm使用JSON消息通过stdin/stdout来和这些subprocess通信。这个通信协议是一个只有100行的库， storm团队给这些库开发了对应的Ruby, Python和Fancy版本。

下面是WordCountTopology里面的SplitSentence的定义:

public static class SplitSentence extends ShellBolt implements IRichBolt {    public SplitSentence() {        super("python","splitsentence.py");    }     public void declareOutputFields(OutputFieldsDeclarer declarer) {        declarer.declare(new Fields("word"));    }}

SplitSentence继承自ShellBolt并且声明这个Bolt用python来运行，并且参数是: splitsentence.py。下面是splitsentence.py的定义:

import storm class SplitSentenceBolt(storm.BasicBolt):    def process(self, tup):        words=tup.values[0].split(" ")        for word in words:          storm.emit([word]) SplitSentenceBolt().run()

8、可靠的消息处理

在这个教程的前面，我们跳过了有关tuple的一些特征。这些特征就是storm的可靠性API： storm如何保证spout发出的每一个tuple都被完整处理。看看以更深入了解storm的可靠性API.

Storm允许用户在Spout中发射一个新的源Tuple时为其指定一个MessageId，这个MessageId可以是任意的Object对象。多个源Tuple可以共用同一个MessageId，表示这多个源Tuple对用户来说是同一个消息单元。Storm的可靠性是指Storm会告知用户每一个消息单元是否在一个指定的时间内被完全处理。完全处理的意思是该MessageId绑定的源Tuple以及由该源Tuple衍生的所有Tuple都经过了Topology中每一个应该到达的Bolt的处理。

在Spout中由message 1绑定的tuple1和tuple2分别经过bolt1和bolt2的处理，然后生成了两个新的Tuple，并最终流向了bolt3。当bolt3处理完之后，称message 1被完全处理了。

Storm中的每一个Topology中都包含有一个Acker组件。Acker组件的任务就是跟踪从Spout中流出的每一个messageId所绑定的Tuple树中的所有Tuple的处理情况。如果在用户设置的最大超时时间内这些Tuple没有被完全处理，那么Acker会告诉Spout该消息处理失败，相反则会告知Spout该消息处理成功。

那么Acker是如何记录Tuple的处理结果呢？？

A xor A = 0.

A xor B…xor B xor A = 0，其中每一个操作数出现且仅出现两次。

在Spout中，Storm系统会为用户指定的MessageId生成一个对应的64位的整数，作为整个Tuple Tree的RootId。RootId会被传递给Acker以及后续的Bolt来作为该消息单元的唯一标识。同时，无论Spout还是Bolt每次新生成一个Tuple时，都会赋予该Tuple一个唯一的64位整数的Id。

当Spout发射完某个MessageId对应的源Tuple之后，它会告诉Acker自己发射的RootId以及生成的那些源Tuple的Id。而当Bolt处理完一个输入Tuple并产生出新的Tuple时，也会告知Acker自己处理的输入Tuple的Id以及新生成的那些Tuple的Id。Acker只需要对这些Id进行异或运算，就能判断出该RootId对应的消息单元是否成功处理完成了。

摘抄自：