Storm是一个分布式的、高容错的实时计算系统。Storm适用的场景：
- Storm可以用来用来处理源源不断的消息，并将处理之后的结果保存到持久化介质中。
- 由于Storm的处理组件都是分布式的，而且处理延迟都极低，所以可以Storm可以做为一个通用的分布式RPC框架来使用。(实时计算？)
Storm集群架构
Storm集群采用主从架构方式，主节点是Nimbus，从节点是Supervisor，有关调度相关的信息存储到ZooKeeper集群中，架构如下图所示
- Nimbus：Storm集群的Master节点，负责分发用户代码，指派给具体的Supervisor节点上的Worker节点，去运行Topology对应的组件（Spout/Bolt）的Task。
- Supervisor：Storm集群的从节点，负责管理运行在Supervisor节点上的每一个Worker进程的启动和终止。
- ZooKeeper：
- 存储客户端提供的topology任务信息，nimbus负责将任务分配信息写入Zookeeper，supervisor从Zookeeper上读取任务分配信息
- 存储supervisor和worker的心跳（包括它们的状态），使得nimbus可以监控整个集群的状态， 从而重启一些挂掉的worker
- 存储整个集群的所有状态信息和配置信息。
组件抽象
我们先看一下，Topology提交到Storm集群后的运行时部署分布图，如下图所示：
通过上图我们可以看出，一个Topology的Spout/Bolt对应的多个Task可能分布在多个Supervisor的多个Worker内部。而每个Worker内部又存在多个Executor，根据实际对Topology的配置在运行时进行计算并分配。
- Topology：Storm对一个分布式计算应用程序的抽象，目的是通过一个实现Topology能够完整地完成一件事情（从业务角度来看）。一个Topology是由一组静态程序组件（Spout/Bolt）、组件关系Streaming Groups这两部分组成。
- Spout：描述了数据是如何从外部系统（或者组件内部直接产生）进入到Storm集群
- Bolt：描述了与业务相关的处理逻辑。
- Task：Spout/Bolt在运行时所表现出来的实体，都称为Task(多个)
- Worker：运行时Task所在的一级容器，Executor运行于Worker中，一个Worker对应于Supervisor上创建的一个JVM实例(和Spark一样的概念)
- Executor：运行时Task所在的直接容器，在Executor中执行Task的处理逻辑
数据流
storm是流式计算框架，数据源源不断地到来。storm中，每条消息称为元组，我们可以把这条消息灵活地看作KV结构。
容错
一般而言，数据在节点间的传递次数氛围以下三种，storm根据用户指定，选择这三种类型的数据保证：
- 至少一次：节点收到相同的数据一次或者多次
- 至多一次：节点最多收到一次数据(S4)，没有容错
- 只有一次：计算正确性的必须保证(strom特有的可以选择数据只被计算一次，防止有些有状态的任务，多次计算后出错)
至少送达一次
storm在保证数据被传递到所有节点方面做得非常巧妙。
- 对于每条消息i，赋予64bit长度的ID。同时，在一张表T(表是逻辑上的，通过一致性hash来找到对应的消息)上记录这条消息的初始ID di = ID。
- 在每个后继节点上，每产生一条消息，就会生成一个随机ID。然后更新di = di ^ 消息输入ID ^ (消息输出ID1 ^ 消息输出ID2 ……)
- 当di = 0，说明已经成功处理消息。
- storm定期扫描T，对于没有成功更新为0的消息，重新发送。对于这种机制，我们可以发现，存在误判的概率2^(-64)
上面说得有点抽象，举例：
上面图12-10不对，系统表T是这样的
| header 1 | header 2 |
|---|---|
| d1 | [11011->0] |
| d2 | [11101->0] |
原理：每一条消息ID，我们都会异或两次，一次是前驱节点在产生这条消息的时候，一次是在后继节点消费这条消息后。
Trident
通过Trident API同时实现了状态持久化和“恰好送达一次”
- 将多条数据封装为一个batch(这不成批处理了吗)，每个batch都有一个递增的ID(要有一个全局递增ID，需要zookeeper的帮助啊)。通过前面的至少发送一次机制，发现某个ID的batch处理失败，重新发送这个batch，ID不变
- 各个计算节点在计算过程中通过ID提交状态，如果发现已经有这个ID的状态，则放弃本次计算。
- ID必须顺序提交
storm高可用性(HA)
- 如果woker挂了，supervisor会重新创建
- bolt任务失败直接重新启动
- Spout任务失败，会再次从数据源拿数据
- 如果机器节点挂了，nimbus会把该节点上的task转移到其他节点
- 如果nimbus或者supervisor挂了，重启就行了。nimbus和supervisor被设计成无状态，- 状态都被存到zookeeper里面了
- 为防止nimbus挂掉，worker节点也挂掉，导致任务无法被nimbus转移到其他机器。nimbus也被设计成HA的，利用主从结构保证主节点挂了之后从节点一样能服务
Stream Groupings(shuffle)
Storm中最重要的抽象，应该就是Stream grouping了，它能够控制Spot/Bolt对应的Task以什么样的方式来分发Tuple，将Tuple发射到目的Spot/Bolt对应的Task，如下图所示：
- Shuffle Grouping：随机分组，跨多个Bolt的Task，能够随机使得每个Bolt的Task接收到大致相同数目的Tuple，但是Tuple不重复
- Fields Grouping：根据指定的Field进行分组 ，同一个Field的值一定会被发射到同一个Task上
- Partial Key Grouping：与Fields grouping 类似，根据指定的Field的一部分进行分组分发
- All Grouping：所有Bolt的Task都接收同一个Tuple（这里有复制的含义）
- Global Grouping：所有的流都指向一个Bolt的同一个Task（也就是Task ID最小的）
- None Grouping：不需要关心Stream如何分组，等价于Shuffle grouping
- Direct Grouping：由Tupe的生产者来决定发送给下游的哪一个Bolt的Task ，这个要在实际开发编写Bolt代码的逻辑中进行精确控制
- Local or Shuffle Grouping：如果目标Bolt有1个或多个Task都在同一个Worker进程对应的JVM实例中，则Tuple只发送给这些Task
最常用的应该是Shuffle Grouping(随机)、Fields Grouping(哈希)、Direct Grouping(用户指定)这三种
Topology并行度计算
conf.setNumWorkers(2); // 该Topology运行在Supervisor节点的2个Worker进程中
topologyBuilder.setSpout("blue-spout", new BlueSpout(), 2); // 设置并行度为2，则Task个数为2*1
topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2)
.setNumTasks(4)
.shuffleGrouping("blue-spout"); // 设置并行度为2，设置Task个数为4 ，则Task个数为4
topologyBuilder.setBolt("yellow-bolt", new YellowBolt(), 6)
.shuffleGrouping("green-bolt"); // 设置并行度为6，则Task个数为6*1
一共有12个任务，10的并行度，2个work。所以一个work里面有5个executor，6个task(分别是1蓝，2绿，3黄)，storm会把同类型的Task尽量放到同一个Executor中运行，Task个数最少的开始分配。
work和excutor内部构造
- 同一work不同executor的两个task消息传递
- 不同work不同executor的两个task消息传递
上图我们还能看出同一executor不同task之间的消息传递
总结：
- 同一work之内的消息传递都要通过executor的消息收发线程
- 不同work的消息传递要通过work的收发线程
- 每个Executor应该维护Task与所在的Executor之间的关系，这样才能正确地将Tuple传输到目的Bolt Task进行处理。