storm 架构原理

参考连接：https://blog.csdn.net/u013332124/article/details/79682782

Storm 是一个分布式的，可靠的，容错的数据流处理系统。下面我将分别从storm的总体架构以及部分原理进行讲解。

1、基本的概念

storm中服务器节点分为主节点和从节点,Nimbus为主节点和Supervisor为从节点。以及若干组件构成。下面为对一些术语进行简单的介绍: 
Nimbus：主节点，是一个调度中心，负责分发任务 
Supervisor：从节点，任务执行的地方 
Worker：任务工做进程，一个Supervisor中能够有多个Worker。 
Executor：Worker进程在执行任务时，会启动多个Executor线程 
Topology：任务的抽象概念。因为storm是流式计算的框架，它的数据流和拓扑图很像，因此它的任务就叫topology。 
Spout：从数据源获取数据并进行分发。 
Bolt：获得Spout或者上一个Bolt的数据,而后进行处理后交给下一个Bolt处理。 
Tuple：在storm中，一条数据能够理解为是一个Tuple。

2、storm的架构

任务提交处理流程

Nimbus是调度中心，Supervisor是任务执行的地方。Supervisor上面有若干个Worker，每一个Worker都有本身的端口,Worker能够理解为一个进程。另外，每一个Worker中还能够运行若干个线程。

当客户端向storm集群提交一个Topology时，这里的提交就是在集群上经过命令storm jar xxx启动topology。若是咱们是在Supervisor节点上执行storm jar xxx，那么Supervisor会将jar包拷贝到Nimbus,以后Nimbus对Topology进行调度。

Nimbus会根据Topology所须要的Worker进行分配，将其分配到各个Supervisor的节点上执行。

如今假设咱们咱们有4个Supervisor节点，每一个Supervisor都配置4个Worker。这是咱们提交了一个Topology，须要4个Worker,那可能的分配状况可能以下图所示: 

storm中的数据流

启动完Topology后,相关组件就开始运行起来了。在Storm中，Spout组件主要用来从数据源拉取数据,造成一个Tuple后转交给Bolt处理。Bolt接受到Tuple处理完后,能够选择继续交给下一个Bolt处理，也能够选择不往下传。这样数据以Tuple的形式一个接一个的往下执行,就造成了一个拓扑数据流。

storm数据在组件间的流向以下图所示: 

3、Storm的并发度

在Storm中，Worker不是组件执行的最小单位。Executor才是，Executor能够理解为是一个线程。咱们在建立topology的时候，能够设置执行spout的线程数和bolt的线程数。

假设spout和bolt的线程数加起来设置了8个，而后设置了2个worker，那么这8个线程可能就会随机分配到2个worker中，可能一个worker3个，一个worker5个。也有可能各自分配4个。以下图所示: 

4、数据的Grouping策略

在实际应用中,Bolt组件的实例可能有多个,Tuple在流向Bolt时，选择哪一个Bolt实例的策略就是grouping策略。 
下面是Storm中的6种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是同样的效果，不平均分配。 
6. Direct Grouping: 直接分组， 这是一种比较特别的分组方法，用这种分组意味着消息的发送者举鼎由消息接收者的哪一个task处理这个消息。 只有被声明为Direct Stream的消息流能够声明这种分组方法。并且这种消息tuple必须使用emitDirect方法来发射。消息处理者能够经过TopologyContext来或者处理它的消息的taskid(OutputCollector.emit方法也会返回taskid)

5、消息的可靠性保证 —— ack机制

一条数据在Spout中造成一个Tuple，而后交给一个个Bolt执行,那咱们怎么保证这个Tuple被完整的执行了呢？这里的完整执行说的是这个Tuple必须在后面的每个Bolt都成功处理，假设在一个Bolt中发生异常致使失败，这就不能算完整处理。

为了保证消息处理过程当中的可靠性,storm使用了ack机制。storm会专门启动若干acker线程，来追踪tuple的处理过程。acker线程数量能够设置。

每个Tuple在Spout中生成的时候,都会分配到一个64位的messageId。经过对messageId进行哈希咱们能够执行要对哪一个acker线程发送消息来通知它监听这个Tuple。

acker线程收到消息后,会将发出消息的Spout和那个messageId绑定起来。而后开始跟踪该tuple的处理流程。若是这个tuple所有都处理完，那么acker线程就会调用发起这个tuple的那个spout实例的ack()方法。若是超过必定时间这个tuple还没处理完，那么acker线程就会调用对应spout的fail()方法,通知spout消息处理失败。spout组件就能够从新发送这个tuple。

从上面的介绍咱们知道了，tuple数据的流向会造成一个拓扑图，也能够理解成是一个tuple树。这个拓扑图的节点可能会有不少个，若是要把这些节点所有保存起来，处理大量的数据时势必会形成内存溢出。

对于这个难题，storm使用了一种很是巧妙的方法，使用20个字节就能够追踪一个tuple是否被完整的执行。这也是storm的一个突破性的技术。

ack机制的具体原理

咱们都知道,本身异或本身,结果确定为零( a ^ a = 0)。ack中就利用这个特性

• acker对于每一个spout-tuple保存一个ack-val的校验值，它的初始值是0， 而后每发射一个tuple/ack一个tuple，那么tuple的id都要跟这个校验值异或一下。注意，这里的tuple的id不是spout-tuple的id,和咱们上面理解的messageId不是一个概念，要区分一下,是每一个新生产的tuple的id，这个tupleId是随机生成的64位比特值
• 以后把获得的值更新为ack-val的新值。那么假设每一个发射出去的tuple都被ack了， 那么最后ack-val必定是0(由于一个数字跟本身异或获得的值是0)。

举个例子,好比发射了某个tuple，就 ack-val ^ tupleId，而后ack了某个tuple,就再ack-val ^ tupleId，这样，ack-val 最终又变成了0，说明tuple已经所有处理成功了。

6、Storm的HA保证——高可用性保证

1. 数据方面的高可用

使用ack机制保证数据处理的高可用

2. Worker进程挂了怎么办？

Supervisor会自动重启worker线程。

3. Supervisor节点失效了怎么办？

能够在其余节点重启该supervisor任务。

4. Nimbus挂了怎么办？

在storm1.0以前,Nimbus是不支持HA的。Nimbus若是挂了，重启Nimbus进程就能够了，不会影响到现有topology的运行。

由于Nimbus只是一个调度中心，Nimbus和Supervisor的状态都保存在本地文件和ZooKeeper，所以他们进程能够随便杀死，而后重启，不会影响到Worker进程的运行。

另外，Nimbus的做用在就是在拓扑任务开始阶段，负责将任务提交到集群，后期负责拓扑任务的管理，好比任务查看，终止等操做。在一般状况下，nimbus的任务压力并不会很大，在天然状况下不会出现宕机的状况。

storm1.0后Nimbus的HA策略尚未具体研究过，有兴趣的小伙伴可自行前往官网查看文档。http://storm.apache.org/releases/1.2.1/nimbus-ha-design.html

7、总结

Storm的架构及原理总体理解起来不算很难，但不少细节仍是须要在实践中才能发现。有兴趣的小伙伴能够去读读storm的源码，storm源码大多数都是用Clojure实现，对Clojure语言不熟悉的朋友能够去看一下JStorm的源码实现。这是阿里基于Storm用java实现的框架，听说更加稳定高效。