概述

 公司之前使用了strom框架来进行实时计算，现在总结一下之前的知识和经验，如有不足之处，望广大网友及时指正，不胜感激。

简而言之：Storm是一个分布式的，可靠的，容错的数据流处理系统。Storm集群的输入流由一个被称作spout的组件管理，spout把数据传递给bolt， bolt要么把数据保存到某种存储器，要么把数据传递给其它的bolt。一个Storm集群就是在一连串的bolt之间转换spout传过来的数据。

STORM组件

在Storm集群中，有两类节点：主节点master node和工作节点worker nodes。主节点运行 Nimbus守护进程，这个守护进程负责在集群中分发代码，为工作节点分配任务，并监控故障。Supervisor守护进程作为拓扑的一部分运行在工作节 点上。一个Storm拓扑结构在不同的机器上运行着众多的工作节点。每个工作节点都是topology中一个子集的实现。而Nimbus和 Supervisor之间的协调则通过Zookeeper系统或者集群。

zookeeper：

Zookeeper是完成Supervisor和Nimbus之间协调的服务。而应用程序实现实时的逻辑则被封装进Storm中的“topology”。 topology则是一组由Spouts（数据源）和Bolts（数据操作）通过Stream Groupings进行连接的图。

spout：

Spout从来源处读取数据并放入topology。Spout分成可靠和不可靠两种；当Storm接收失败时，可靠的Spout会对tuple（元组， 数据项组成的列表）进行重发；而不可靠的Spout不会考虑接收成功与否只发射一次。而Spout中最主要的方法就是nextTuple（），该方法会发 射一个新的tuple到topology，如果没有新tuple发射则会简单的返回。

bolt：

Topology中所有的处理都由Bolt完成。Bolt从Spout中接收数据并进行处理，如果遇到复杂流的处理也可能将tuple发送给另一个 Bolt进行处理。而Bolt中最重要的方法是execute（），以新的tuple作为参数接收。不管是Spout还是Bolt，如果将tuple发射 成多个流，这些流都可以通过declareStream（）来声明。

Stream Groupings：

Stream Grouping定义了一个流在Bolt任务中如何被切分。

1. Shuffle grouping：随机分发tuple到Bolt的任务，保证每个任务获得相等数量的tuple。

2.Fields grouping：根据指定字段分割数据流，并分组。例如，根据“user-id”字段，相同“user-id”的元组总是分发到同一个任务，不同“user-id”的元组可能分发到不同的任务。

3. Partial Key grouping：根据指定字段分割数据流，并分组。类似Fields grouping。

4.All grouping：tuple被复制到bolt的所有任务。这种类型需要谨慎使用。

5. Global grouping：全部流都分配到bolt的同一个任务。明确地说，是分配给ID最小的那个task。

6. None grouping：无需关心流是如何分组。目前，无分组等效于随机分组。但最终，Storm将把无分组的Bolts放到Bolts或Spouts订阅它们的同一线程去执行（如果可能）。

7. Direct grouping：这是一个特别的分组类型。元组生产者决定tuple由哪个元组处理者任务接收。

8. Local or shuffle grouping：如果目标bolt有一个或多个任务在同一工作进程，tuples 会打乱这些进程内的任务。否则,这就像一个正常的 Shuffle grouping。

STORM原理

Storm称用户的一个作业为Topology（拓扑），为什么叫拓扑呢？是因为Storm的一个拓扑主要包含了许多的数据节点，还有一些计算节点，以及这些节点之间的边，也就是说Storm的拓扑是由这些点和边组成的一个有向无环图。这些点有两种：数据源节点（Spout）、普通的计算节点（Bolt），点之间的边称为数据流（Stream），数据流中的每一条记录称为Tuple。

如下图中，每一个“水龙头”表示一个Spout，它会发送一些Tuple给下游的Bolt，这些Bolt经过处理周，再发送一个Tuple给下一个Bolt，最后，在这些Bolt里面是可以执行一些写数据到外部存储（如数据库）等操作的。在图中这个Topology里面我们看到了两个Spout和5个Bolt，在实际运行的时候，每个Spout节点都可能有很多个实例，每个Bolt也有可能有很多个实例。就像MapReduce一样，一个Map节点并不代表只有一个并发，而有可能很多个Map实例在跑。

这些Spout和Bolt的这些边里面，用户可以设置多种的Grouping的方式。有些类似SQL中的Group By。用来制定这些计算是怎么分组的。

 *Fields Grouping：保证同样的字段移动落到同一个Bolt里，

Topologies

为了在storm上面做实时计算， 你要去建立一些topologies。一个topology就是一个计算节点所组成的图。Topology里面的每个处理节点都包含处理逻辑， 而节点之间的连接则表示数据流动的方向。运行一个Topology是很简单的。首先，把你所有的代码以及所依赖的jar打进一个jar包。然后运行类似下面的这个命令。strom jar all-your-code.jar backtype.storm.MyTopology arg1 arg2这个命令会运行主类: backtype.strom.MyTopology，参数是arg1, arg2。这个类的main函数定义这个topology并且把它提交给Nimbus。storm jar负责连接到nimbus并且上传jar文件。

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里面， 你可以指定每个节点的并行度， storm则会在集群里面分配那么多线程来同时计算。

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

数据模型(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。

 1     publicclassDoubleAndTripleBoltimplementsIRichBolt {
 2         privateOutputCollectorBase _collector;
 3 
 4         @Override
 5         publicvoidprepare(Map conf, TopologyContext context, OutputCollectorBase collector) {
 6             _collector = collector;
 7         }
 8 
 9         @Override
10         publicvoidexecute(Tuple input) {
11             intval = input.getInteger(0);
12             _collector.emit(input,newValues(val*2, val*3));
13             _collector.ack(input);
14         }
15 
16         @Override
17         publicvoidcleanup() {
18         }
19 
20         @Override
21         publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {
22             declarer.declare(newFields("double","triple"));
23         }
24     }

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

STORM和其他工具对比

目前比较流行的实时处理引擎有 Storm，Spark Streaming，Flink。每个引擎都有各自的特点和应用场景。 下表是对这三个引擎的简单对比。

 

 

 strom安装比较简单，这里省略安装步骤，安装成功如图所示：

STORM原理

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

Task数量：表示每个Spout或Bolt逻辑上有多少个并发。它影响输出结果。

Worker数量：代表总共有几个JVM进程去执行我们的作业。

Executor数量：表示每个Spout或Bolt启动几个线程来运行

 下面代码中的数字表示Executor数量，它不影响结果，影响性能。

Worker的数量在Config中设置，下图代码中的部分表示Worker数量。

*本地模式中，Worker数不生效，只会启动一个JVM进行来执行作业。

*只有在集群模式设置Worker才有效。而且集群模式的时候一定要设置才能体现集群的价值。

 数据可靠性：

（1）Spout容错API：NextTuple中，emit时，指定MsgID。

（2）Bolt容错API：①emit时，锚定输入Tuple。②Act输入Tuple。

STORM参数设置

storm.zookeeper.servers：

ZooKeeper服务器列表

storm.zookeeper.port：

ZooKeeper连接端口

storm.local.dir：

storm使用的本地文件系统目录(必须存在并且storm进程可读写)

storm.cluster.mode：

Storm集群运行模式([distributed|local])

storm.local.mode.zmq：

Local模式下是否使用ZeroMQ作消息系统，如果设置为false则使用java消息系统。默认为false

storm.zookeeper.root：

ZooKeeper中Storm的根目录位置

storm.zookeeper.session.timeout：

客户端连接ZooKeeper超时时间

storm.id：

运行中拓扑的id，由storm name和一个唯一随机数组成。

nimbus.host：

nimbus服务器地址

nimbus.thrift.port：nimbus的thrift监听端口

nimbus.childopts：

通过storm-deploy项目部署时指定给nimbus进程的jvm选项

nimbus.task.timeout.secs：

心跳超时时间，超时后nimbus会认为task死掉并重分配给另一个地址

nimbus.monitor.freq.secs：

nimbus检查心跳和重分配任务的时间间隔。注意如果是机器宕掉nimbus会立即接管并处理

nimbus.supervisor.timeout.secs：

supervisor的心跳超时时间，一旦超过nimbus会认为该supervisor已死并停止为它分发新任务

nimbus.task.launch.secs：

task启动时的一个特殊超时设置。在启动后第一次心跳前会使用该值来临时替代nimbus.task.timeout.secs

nimbus.reassign：

当发现task失败时nimbus是否重新分配执行。默认为真，不建议修改

nimbus.file.copy.expiration.secs：

nimbus判断上传/下载链接的超时时间，当空闲时间超过该设定时nimbus会认为链接死掉并主动断开

ui.port：

Storm UI的服务端口

drpc.servers：

DRPC服务器列表，以便DRPCSpout知道和谁通讯

drpc.port：

Storm DRPC的服务端口

supervisor.slots.ports：

supervisor上能够运行workers的端口列表。每个worker占用一个端口，且每个端口只运行一个worker。

通过这项配置可以调整每台机器上运行的worker数。(调整slot数/每机)

supervisor.childopts：

在storm-deploy项目中使用，用来配置supervisor守护进程的jvm选项

supervisor.worker.timeout.secs：

supervisor中的worker心跳超时时间，一旦超时supervisor会尝试重启worker进程.

supervisor.worker.start.timeout.secs：

supervisor初始启动时，worker的心跳超时时间，当超过该时间supervisor会尝试重启worker。

因为JVM初始启动和配置会带来的额外消耗，从而使得第一次心跳会超过supervisor.worker.timeout.secs的设定

supervisor.enable：

supervisor是否应当运行分配给他的workers。默认为true，该选项用来进行Storm的单元测试，一般不应修改.

supervisor.heartbeat.frequency.secs：

supervisor心跳发送频率(多久发送一次)

supervisor.monitor.frequency.secs：

supervisor检查worker心跳的频率

worker.childopts：

supervisor启动worker时使用的jvm选项。所有的”%ID%”字串会被替换为对应worker的标识符

worker.heartbeat.frequency.secs：

worker的心跳发送时间间隔

task.heartbeat.frequency.secs：

task汇报状态心跳时间间隔

task.refresh.poll.secs：

task与其他tasks之间链接同步的频率。(如果task被重分配，其他tasks向它发送消息需要刷新连接)

。一般来讲，重分配发生时其他tasks会理解得到通知。该配置仅仅为了防止未通知的情况。

topology.debug：

如果设置成true，Storm将记录发射的每条信息。

topology.optimize：

master是否在合适时机通过在单个线程内运行多个task以达到优化topologies的目的

topology.workers：

执行该topology集群中应当启动的进程数量。

每个进程内部将以线程方式执行一定数目的tasks。topology的组件结合该参数和并行度提示来优化性能

topology.ackers：

topology中启动的acker任务数。

Acker保存由spout发送的tuples的记录，并探测tuple何时被完全处理。

当Acker探测到tuple被处理完毕时会向spout发送确认信息。通常应当根据topology的吞吐量来确定acker的数目，但一般不需要太多。

当设置为0时，相当于禁用了消息可靠性。storm会在spout发送tuples后立即进行确认

topology.message.timeout.secs：

topology中spout发送消息的最大处理超时时间。

如果一条消息在该时间窗口内未被成功ack，Storm会告知spout这条消息失败。而部分spout实现了失败消息重播功能。

topology.kryo.register：

注册到Kryo(Storm底层的序列化框架)的序列化方案列表。序列化方案可以是一个类名，或者是com.esotericsoftware.kryo.Serializer的实现

topology.skip.missing.kryo.registrations：

Storm是否应该跳过它不能识别的kryo序列化方案。如果设置为否task可能会装载失败或者在运行时抛出错误

topology.max.task.parallelism：

在一个topology中能够允许的最大组件并行度。该项配置主要用在本地模式中测试线程数限制.

topology.max.spout.pending：

一个spout task中处于pending状态的最大的tuples数量。该配置应用于单个task，而不是整个spouts或topology

topology.state.synchronization.timeout.secs：

组件同步状态源的最大超时时间(保留选项,暂未使用)

topology.stats.sample.rate：

用来产生task统计信息的tuples抽样百分比

topology.fall.back.on.java.serialization：

topology中是否使用java的序列化方案

zmq.threads：

每个worker进程内zeromq通讯用到的线程数

zmq.linger.millis：

当连接关闭时，链接尝试重新发送消息到目标主机的持续时长。这是一个不常用的高级选项，基本上可以忽略.

java.library.path：

JVM启动(如Nimbus,Supervisor和workers)时的java.library.path设置。该选项告诉JVM在哪些路径下定位本地库。

STORM一个简单例子

先看一个简单例子：

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

这个Topology包含一个Spout和两个Bolt。Spout发射单词。这三个节点被排成一条线: spout发射单词给第一个bolt， 第一个bolt然后把处理好的单词发射给第二个bolt。

我们使用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。

STORM测试代码

没有代码的技术文档总感觉少些说服性，废话不说，测试代码，使用了storm中example的代码中的kafka部分，提交命令成功截图是：

 

 从上图可知一共提交了4个拓扑，现在在UI界面 讲解一下每个页面的作用和说明：

1，首页运行界面可以看到Topology的概述，包括拓扑name，状态，：

Name: topology name

id: topology id (由storm生成)

status: topology的状态，包括(ACTIVE, INACTIVE, KILLED, REBALANCING)

uptime: topology运行的时间

num workers: 运行的workers数

num tasks: 运行的task数

2，进入其中一个拓扑展示界面是：

【topology stats】

window: 时间窗口，显示10m、3h、1d和all time的运行状况

emitted: emitted tuple数

transferred: transferred tuple数, 说下与emitted的区别：如果一个task，emitted一个tuple到2个task中，则transferred tuple数是emitted tuple数的两倍

complete latency: spout emitting 一个tuple到spout ack这个tuple的平均时间

acked: ack tuple数

failed: 失败的tuple数

【spouts】

id: spout id

parallelism: 任务数

last error: 最近的错误数，只显示最近的前200个错误

emitted、transferred、complete latency、acked和failed上面已解释

【bolts】

process latency: bolt收到一个tuple到bolt ack这个tuple的平均时间

其他参数都解释过了

还有componentpage和taskpage，参数的解释同上。

taskpage中的Component指的是spoutid或者boltid，time指的是错误发生的时间，error是指错误的具体内容。

STORM优化

kryo序列化

定制序列化

自定义的bolt之间emit数据是实体类的时候，注册kryo

Storm 使用 Kryo 来处理序列化。如果要实现自定义的序列化生成器，需要注册一个新的 Kryo 的序列化生成器。

可以通过拓扑的 topology.kryo.register 属性来添加自定义序列化生成器。两种姿势：

只有一个待注册的类的名称。在这种情况下，Storm 会使用 Kryo 的 FieldsSerializer 来序列化该类。conf.registerSerialization(UserEntity.class);

一个包含待注册的类的名称和对应的序列化实现类名称，该序列化实现类实现了 com.esotericsoftware.kryo.Serializer接口。

topology.kryo.register:
  - com.ly.CustomType1
  - com.ly.CustomType2: com.ly.serializer.CustomType2Serializer
  - com.ly.CustomType3

KafkaBolt批量提交

1 props.put("request.required.acks", "0");
2 props.put("producer.type", " async");
3 props.put("message.send.max.retries","3");
4 props.put("batch.num.messages", "200");
5 props.put("send.buffer.bytes", "1024*100");

使用组件的并行度代替线程池

Storm 自身是一个分布式、多线程的框架，对每个Spout 和Bolt，我们都可以设置其并发度；它也支持通过rebalance 命令来动态调整并发度，把负载分摊到多个Worker 上。如果自己在组件内部采用线程池做一些计算密集型的任务，比如JSON 解析，有可能使得某些组件的资源消耗特别高，其他组件又很低，导致Worker 之间资源消耗不均衡，这种情况在组件并行度比较低的时候更明显。
如某个Bolt 设置了1 个并行度，但在Bolt 中又启动了线程池，这样导致的一种后果就是，集群中分配了这个Bolt 的Worker 进程可能会把机器的资源都给消耗光了，影响到其他Topology 在这台机器上的任务的运行。如果真有计算密集型的任务，我们可以把组件的并发度设大，Worker 的数量也相应提高，让计算分配到多个节点上。

注意fieldsGrouping 的数据均衡性

fieldsGrouping 是根据一个或者多个Field 对数据进行分组，不同的目标Task 收到不同的数据，而同一个Task 收到的数据会相同。
假设某个Bolt 根据用户ID 对数据进行fieldsGrouping，如果某一些用户的数据特别多，而另外一些用户的数据又比较少，那么就可能使得下一级处理Bolt 收到的数据不均衡，整个处理的性能就会受制于某些数据量大的节点。可以加入更多的分组条件或者更换分组策略，使得数据具有均衡性。

优先使用localOrShuffleGrouping

localOrShuffleGrouping 是指如果目标Bolt 中的一个或者多个Task 和当前产生数据的Task 在同一个Worker 进程里面，那么就走内部的线程间通信，将Tuple 直接发给在当前Worker 进程的目的Task。否则，同shuffleGrouping。localOrShuffleGrouping 的数据传输性能优于shuffleGrouping，因为在Worker 内部传输，只需要通过Disruptor 队列就可以完成，没有网络开销和序列化开销。因此在数据处理的复杂度不高， 而网络开销和序列化开销占主要地位的情况下，可以优先使用localOrShuffleGrouping 来代替shuffleGrouping。

设置合理的Worker 数

Worker是运行在工作节点上面，被Supervisor守护进程创建的用来干活的JVM进程。每个Worker对应于一个给定topology的全部执行任务的一个子集。反过来说，一个Worker里面不会运行属于不同的topology的执行任务。它可以通过[storm rebalance]命令任意调整。

Worker 数越多，性能越好？先看一张Worker 数量和吞吐量对比的曲线

从图可以看出，在12 个Worker 的情况下，吞吐量最大，整体性能最优。这是由于一方面，每新增加一个Worker 进程，都会将一些原本线程间的内存通信变为进程间的网络通信，这些进程间的网络通信还需要进行序列化与反序列化操作，这些降低了吞吐率。
另一方面，每新增加一个Worker 进程，都会额外地增加多个线程（Netty 发送和接收线程、心跳线程、SystemBolt 线程以及其他系统组件对应的线程等），这些线程切换消耗了不少CPU，sys 系统CPU 消耗占比增加，在CPU 总使用率受限的情况下，降低了业务线程的使用效率。

设置合理的Executor数

可以理解成一个Worker进程中的工作线程。

一个Executor中只能运行隶属于同一个component（spout/bolt）的task。

一个Worker进程中可以有一个或多个Executor线程。在默认情况下，一个Executor运行一个task。

每个component（spout/bolt）的并发度就是指executor数量。

它可以通过[storm rebalance]命令任意调整。

设置合理的Task数

Task则是spout和bolt中具体要干的活了。一个Executor可以负责1个或多个task。

同时，task也是各个节点之间进行grouping（partition）的单位。无法在运行时调整。

--设置方法：

conf.setNumWorkers(workers);                                        //设置worker数量

uilder.setBolt("2", new WordSpliter(),4)                             //设置Executor并发数量

builder.setBolt("2", new WordSpliter(),4).setNumTasks(1); //设置每个线程处理的Task数量

--任务分配：

任务分配是有下面两种情况：

①、task数目比worker多：

例如task是[1 2 3 4]，可用的slot(所谓slot就是可用的worker)只有[host1:port1，host2:port1]，那么最终是这样分配
1：[host1:port1]

2：[host2:port1]

3：[host1:port1]

4：[host2:port1]

②、task数目比worker少：

例如task是[1 2]，而worker有[host1:port1，host1:port2，host2:port1，host2:port2]，

那么首先会将woker排序，将不同host间隔排列，保证task不会全部分配到同一个机器上，也就是将worker排列成

[host1:port1，host2:port1，host1:port2，host2:port2]

然后分配任务为：

1：[host1:port1]

2：[host2:port1]

总结

感谢网络大神的分享：

https://blog.csdn.net/wangshuminjava/article/details/79367254

https://www.aboutyun.com//forum.php%5C?mod=viewthread&tid=7394&extra=page%3D1&page=1&

https://github.com/quchunhui/StormWordCount/blob/master/src/main/java/WordCountTopology.java

https://www.jianshu.com/p/1d17155c54c9