前言：storm作为流式实时计算的代表之一，它的地位对于流式计算相当于hadoop在离线计算中一般。抛开编码过程，本文就来聊聊storm的构造和工作原理。

一.Storm的核心组件

Nimbus：负责资源分配和任务调度。(充当集群中master的角色)。

Supervisor：负责接受nimbus分配的任务，启动和停止属于自己管理的worker进程。(充当slave角色)

---可以通过配置文件设置当前supervisor上启动多少个worker。

Worker：运行具体处理组件逻辑的进程。Worker运行的任务类型只有两种，一种是Spout任务，一种是Bolt任务。(Worker相当于一个JVM，是一个进程，一个topology可以有一个或多个worker，他们并行跑在多个物理机上，并且一个Worker只能从属一个topology)。

executor：worker中每一个spout/bolt下的线程称为executor。一个Worker可以包含多个executor。

Task：Task是用户定义的处理逻辑对象，每一个Spout/Bolt会被当作很多task在整个集群里面执行。

--默认情况下，task数量设置的和executor相同，即一个executor运行一个task。

zookeeper： Apache的第三方工具。storm使用它来协调主从之间的工作。

 

下面再看看代码中怎么设置

//-----------------------------------这是一个简单的storm主程序------------------------------------

Config conf = newConfig();

conf.setNumWorkers(2); //用2个worker

topologyBuilder.setSpout("blue-spout", newBlueSpout(), 2); //设置2个并发度(一共两个executor，每个executor下运行默认的一个task)

topologyBuilder.setBolt("green-bolt", newGreenBolt(), 2).setNumTasks(4).shuffleGrouping("blue-spout"); //设置2个并发度，4个任务(一共两个executor，四个task，每个executor下运行两个task)

topologyBuilder.setBolt("yellow-bolt", newYellowBolt(), 6).shuffleGrouping("green-bolt"); //设置6个并发度(一共六个executor，每个executor下运行默认的一个task)

StormSubmitter.submitTopology("mytopology", conf, topologyBuilder.createTopology());

 

二.Storm编程模型

Topology：Storm中运行的一个实时应用程序的名称。（称作拓扑）

Spout：在一个topology中获取源数据流的组件。

通常情况下spout会从外部数据源中读取数据，然后转换为topology内部的源数据。

Bolt：接受数据然后执行处理的组件,用户可以在其中执行自己想要的操作。

Tuple：一次消息传递的基本单元，理解为一组消息就是一个Tuple。

Stream：表示数据的流向。定义怎么从一堆task发射tuple到另外一堆task。

 

 

三.Storm间的通信

1.Worker进程间的通信(使用ZeroMQ或Netty(0.9以后默认使用作为进程间通信的消息框架)。

1>、  Worker接受线程通过网络接受数据，并根据Tuple中包含的taskId，匹配到对应的executor；然后根据executor找到对应的incoming-queue，将数据存发送到incoming-queue队列中。

2>、  业务逻辑执行现成消费incoming-queue的数据，通过调用Bolt的execute(xxxx)方法，将Tuple作为参数传输给用户自定义的方法

3>、  业务逻辑执行完毕之后，将计算的中间数据发送给outgoing-queue队列，当outgoing-queue中的tuple达到一定的阀值，executor的发送线程将批量获取outgoing-queue中的tuple,并发送到Worker的transfer-queue中

4>、  Worker发送线程消费transfer-queue中数据，计算Tuple的目的地，连接不同的node+port将数据通过网络传输的方式传送给另一个的Worker。

5>、  另一个worker重复执行操作。

 

2.Worker内部通信采用 Disruptor 技术 本篇就不花费篇幅来介绍该技术了。

 

四.任务提交的过程

五.storm的ack-fail(消息容错)机制

Tip:这里只做简单介绍，具体详情看完后有兴趣可自行了解

1.开发中在spout类中发送数据时，调用方法emit(new Values("haha") , msgId); 时候传入了第二个参数msgId是用于该机制的(若你不想用，不关心是否发送失败，则不传该参数)。

2.此时用户发送的数据"haha"，作为一个tuple在一个或者多个bolt中处理。每一次分发数据都会使用新的tuple，最初从spout中发送的那个tuple，假定称作"根源"。以根源为基点会行程tuple树，在这颗树的任何一个节点发送失败，则会调用spout的fail()方法，全部成功调用ack()方法。

3.值得注意的是，storm并不会帮你重发信息，它只是通知你失败，需要你自己去重发。而且在bolt中需要显示调用collector的ack()方法 告知storm这一步成功了

 

扩展：storm是怎么知道一颗树有个节点失败的呢？

大概算法是这样的，整颗树的任何一个节点发送信息时候，附带这个msgid(记作发送方)，收到信息也会记录一次msgid(记作接收方)。所以整颗树种的msgId是成双成对出现的，把他们依次进行异或计算，得到的最后结果会为0。存入map中，不为0的那一颗树是失败的。