推荐慕课网视频：

 

====Storm的起源。

Storm是开源的、分布式、流式计算系统

 

什么是分布式呢？就是将一个任务拆解给多个计算机去执行，让许多机器共通完成同一个任务，

把这个多机的细节给屏蔽，对外提供同一个接口、同一个服务，这样的系统就是分布式系统。

 

在多年以前并没有非常范用的分布式系统，即使存在，也都是限定在指定的领域，

当然，也有人尝试从中提取出共通的部分，发明一个通用的分布式系统，但是都没有很好的结果。

后来，Google发表了3篇论文，提出了分布式计算的模型，在分布式系统上有了一个质的突破。

 

有一位大牛看了这3篇论文之后，深受启发，然后就发明了Hadoop系统。

 

然后，基于Hadoop的改造系统就如雨后春笋一般，接二连三的出现了。

以至于，Hadoop已经不是一套软件，而是一整套生态系统了。

于是，人们谈到分布式，就必谈Hadoop了。

 

但是，Hadoop并不是万能的，它只能处理适合进行批量计算的需求。对于，非批量的计算就不能够满足要求了。

很多时候，我们只能先收集一段时间数据，等数据收集到一定规模之后，我们才开始MapReduce处理。

 

有这么一个故事：

-------------------

路人甲是在一家媒体公司A工作，他的主要工作内容很简单，就是在一些搜索引擎上做广告，

众所周知，搜索引擎上的广告是竞价排名的，谁土豪谁就排前面，出钱少的就只能排在后面。

公司A的竞争对手都比较土豪，所以呢，公司A的广告就一直排在后面，也没什么好的办法。

后来，路人甲想出了一个馊主意，就是用程序不断的去点击竞争对手的广告，让对手的广告费

很快的花费调，这样公司A就可以廉价的将广告排在前面了。

搜索引擎公司试图识别出这些恶意点击屏来保护商家，将这些恶意点击扣除的费用返还给商家。

一般来说呢，如果利用MapReduce，一般情况下，都需要收集一段时间数据，然后根据这些

数据来算出哪些点击是恶意的，本身收集数据就已经很耗费时间了，再等计算完毕之后，

土豪商家的广告费也基本上不剩什么了。

所以呢，我们希望在点击发生的时候就算出来该点击是否是作弊行为，及时不能马上判断出，

也应该尽早的计算出来。

-------------------

 为了解决上面这个故事的需求，分布式流式计算系统就产生了，比较知名的有：

•【Yahoo】S4

•【IBM】StreamBase

•【Amazon】Kinesis

•【Spark】Streaming

•【Google】Millwheel

•【Apache】Storm（目前业界中最知名、流程）

 

批量计算（以Hadoop为代表）与流式计算的区别有哪些呢？

 

###################

目前已经有人在做一些前瞻性的项目，这些人试图将批量计算和流式计算进行整合

试图使用同一套API，即搞定流式计算，又搞定批量计算。

使一段代码不要任何改动，就可以同时执行在批量计算和流式计算两种系统之上。

这种系统目前比较有名的有：

【Twitter】Summing Bird

【Google】CloudDataflow

两个接口都已经开源了。等以后有机会一定要提前接触一下。

###################

 

====Storm组件

Storm采用的是主从结构，就是使用一台主节点来管理整个集群的运行状态。

这个主节点被称为：Nimbus，从节点用来维护每台机器的状态，被称为：Supervisor

 

为什么采取主从结构呢？主从结构比较简单，不需要进行主节点仲裁等工作。

 

从前面的结构图中我们还可以看出，采取主从结构之后，Nimbus是一个单点，

但是，我们知道分布式领域里，大家都比较讨厌自己的系统设计中存在单点，

因为单点如果发生故障，很有可能影响到整个集群的可用性。

 

所以，如果一个系统设计中如果存在单点，一般情况下这个单点的作业必然比较轻，

挂了之后，短时间之内也不影响真个系统的运行，并且一般情况下都是没有状态的，

宕机之后至需要重启就能够恢复并正确处理。

 

Nimbus的角色是只负责一些管理性的工作，它并不关心Worker之间的数据是如何传输的，

它的一些主要状态都存在分布式协调服务(Zookeeper)中，内存里面的东西都是可以丢失的，

如果它挂掉，只要没有运算节点发生故障，那么整个作业还是能够正常的进行数据处理的。

Nimbus重启之后，就可以正确处理真个系统的事务了。

 

Supervisor的角色是听Nimbus的话，来启动并监控真正进行计算的Worker的进程，

如果Worker有异常，那么久帮助Worker重启一下，它也不负责数据计算和数据传输，

 

真正的数据计算和输出，都是由Worker来进行。

Worker是运行在工作节点上面，被Supervisor守护进程创建的用来干活的JVM进程。

每个Worker对应于一个给定topology的全部执行任务的一个子集。

反过来说，一个Worker里面不会运行属于不同的topology的执行任务。

 

====Storm UI

为了方便用户管理集群，查看集群运行状态，提供了一个基于Web的UI来监控整个Storm集群

它本身不是集群运行的必须部分，它的启动停止都不影响Storm的正常运行。

 

====Storm作业提交运行流程

（1）用户使用Storm的API来编写Storm Topology。

（2）使用Storm的Client将Topology提交给Nimbus。

Nimbus收到之后，会将把这些Topology分配给足够的Supervisor。

（3）Supervisor收到这些Topoligy之后，Nimbus会指派一些Task给这些Supervisor。

（4）Nimvus会指示Supervisor为这些Task生成一些Worker。

（5）Worker来执行这些Task来完成计算任务。

  

====StormAPI基础概念

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里。

 

--以WordCount为例，MapReduce和Storm的工作流程对比：

（1）MapReduce

（2）Storm

 

====各个组件的一些说明

--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. public class DoubleAndTripleBoltimplementsIRichBolt {
2.     private OutputCollectorBase _collector;
4.     @Override
5.     public void prepare(Map conf, TopologyContext context, OutputCollectorBase collector) {
6.         _collector = collector;
7.     }
9.     @Override
10.     public void execute(Tuple input) {
11.         intval = input.getInteger(0);
12.         _collector.emit(input,newValues(val*2, val*3));
13.         _collector.ack(input);
14.     }
16.     @Override
17.     public void cleanup() {
18.     }
20.     @Override
21.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
22.         declarer.declare(newFields("double","triple"));
23.     }
24. }

 

参考博客：

 

====StormAPI使用

我们来看看WorldCount的example代码。

 

====Storm的并发机制

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

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

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

 

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

 

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

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

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

 

====Storm数据可靠性

分布式系统都管理很多台机器，需要保证任意的Worker挂掉之后，我们的系统仍然能正确的处理，那么

Storm如何保证这些数据正确的恢复？

Storm如何保证这些数据不被重复计算？

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

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

 

====Storm集群搭建

（1）安装zookeeper集群

配置方法省略。

 

（2）下载安装Storm

官网上下载Storm：http://storm.apache.org

上传至Linux并解压缩。这里将Storm解压缩到/opt/apache-storm-0.10.0路径下了。

 

（3）修改Storm配置文件

配置文件路径：/opt/apache-storm-0.9.5/conf/storm.yaml

配置内容如下：

----------------

storm.zookeeper.servers:

- "192.168.93.128"

- "192.168.93.129"

- "192.169.93.130"

nimbus.host: "192.168.93.128"

storm.local.dir: "/opt/apache-storm-0.9.5/status"

supervisor.slots.ports:

- 6700

- 6701

- 6702

- 6703

----------------

 

置之后的文件如下如所示：

 

--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在哪些路径下定位本地库

 

（4）配置Storm环境变量

环境变量位置：/etc/profile

配置内容之后如下图所示：

注意：环境变量修改只有，一定要使用Source命令来使之生效。

 

（5）启动Storm

--启动Storm UI

命令：storm ui >/dev/null 2>&1 &

我们可以它启动的时候相关的输出指向到/def/null，并且把错误也重新定向到正常输出。

 

--启动主节点（Nimbus节点）

命令：storm nimbus >/dev/null 2>&1 &

在第1台Linux虚拟机上执行。正常启动时的jps结果如下图所示：

 

--启动工作节点（Supervisor节点）

命令：storm supervisor >/dev/null 2>&1 &

在第2、3台Linux虚拟机上执行。正常启动时的jps结果如下图所示：

 

（6）启动StormUI监控页面：

Storm正常启动之后，应该可以打开StormUI画面。在浏览器中输入地址和端口即可

正确启动时应该如下图所示：

--Mainpage：

main页面主要包括3个部分

 

 【Cluster Summary】

•Nimbus uptime: nimbus的启动时间

•Supervisors: storm集群中supervisor的数目

•used slots: 使用了的slots数

•free slots: 剩余的slots数

•total slots: 总的slots数

•Running tasks: 运行的任务数

 

【topology summary】

•Name: topology name

•id: topology id (由storm生成)

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

•uptime: topology运行的时间

•num workers: 运行的workers数

•num tasks: 运行的task数

 

【supervisor summary】

•host: supervisor(主机)的主机名

•uptime: supervisor启动的时间

•slots: supervisor的端口数

•used slots: 使用的端口数

 

--Topology page

topology页面主要包括4个部分

【topology summary】

(同主页)

 

【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常用命令

【提交Topologies】

命令格式：storm jar 【jar路径】 【拓扑包名.拓扑类名】 【拓扑名称】

样例：storm jar /storm-starter.jar storm.starter.WordCountTopology wordcountTop

#提交storm-starter.jar到远程集群，并启动wordcountTop拓扑。

 

【停止Topologies】

命令格式：storm kill 【拓扑名称】

样例：storm kill wordcountTop

#杀掉wordcountTop拓扑。

 

【启动nimbus后台程序】

命令格式：storm nimbus

 

【启动supervisor后台程序】

命令格式：storm supervisor

 

【启动drpc服务】

命令格式：storm drpc

 

【启动ui服务】

命令格式：storm ui

 

【启动REPL】

REPL — read-evaluate-print-loop。

虽然clojure可以作为一种脚本语言内嵌在java里面，但是它的首选编程方式是使用REPL，这是一个简单的命令行接口，

使用它你可以输入你的命令，执行，然后查看结果， 你可以以下面这个命令来启动REPL：

命令格式：storm repl

 

【打印本地配置】

命令格式：storm localconfvalue [配置参数关键字]

举例：storm localconfvalue storm.zookeeper.servers

#根据指定参数打印本地配置的值。

 

【打印远程配置】

命令格式：storm remoteconfvalue [配置参数关键字]

举例：storm remoteconfvalue storm.zookeeper.servers

#根据指定参数打印远程配置的值。

 

【执行Shell脚本】

命令格式：storm shell resourcesdir command args

 

【打印CLASSPATH】

命令格式：storm classpath

 

====Storm调优：

--调优对象

当一个topology在storm cluster中运行时，它的并发主要跟3个逻辑对象相关：worker => executor =>task。(=>代表1对N)

（1）Worker

Worker是运行在工作节点上面，被Supervisor守护进程创建的用来干活的JVM进程。

每个Worker对应于一个给定topology的全部执行任务的一个子集。

反过来说，一个Worker里面不会运行属于不同的topology的执行任务。

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

（2）Executor

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

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

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

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

（3）Task

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

每个component（spout/bolt）的并发度就是这个component对应的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]

 

--简单举例：

通过Config.setNumWorkers(int))来指定一个storm集群中执行topolgy的进程数量，所有的线程将在这些指定的worker进程中运行。

比如说一个topology中要启动300个线程来运行spout/bolt，而指定的worker进程数量是60个。

那么storm将会给每个worker分配5个线程来跑spout/bolt。

如果要对一个topology进行调优，可以调整worker数量和spout/bolt的parallelism(并发度，即executor)数量。

(调整参数之后要记得重新部署topology，后续会为该操作提供一个swapping的功能来减小重新部署的时间)。

 

例如：

builder.setBolt("cpp", new CppBolt(), 3).setNumTasks(5).noneGrouping(pre_name); 

会创建3个线程，但有内存中会5个CppBolt对象，3个线程调度5个对象。

 

--网上搜罗的一些经验：

①、对于worker和task之间的比例，网上也给出了参考,。即1个worker包含10~15个左右。当然这个参考，实际情况还是要根据配置和测试情况。

②、executor数最大不能超过该bolt的task数。

 

--Strom集群命令

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

[root @h2master  bin]# storm  Commands:           activate           classpath           deactivate           dev-zookeeper           drpc           help  命令帮助           jar    执行上传的jar包           kill    杀死正在执行的topology 后面跟 topology的名称           list    查看运行的所有topology运行情况           localconfvalue           logviewer    启动topology日志           nimbus       启动nimbus           rebalance   shell方式下修改topology运行参数比如worker个数 task个数等           remoteconfvalue           repl           shell           supervisor   启动supervisor           ui               启动topology ui界面           version         Help:            help            help <command>

 

[root@h2master bin]# storm
Commands:
        activate
        classpath
        deactivate
        dev-zookeeper
        drpc
        help 命令帮助
        jar   执行上传的jar包
        kill   杀死正在执行的topology 后面跟 topology的名称
        list   查看运行的所有topology运行情况
        localconfvalue
        logviewer   启动topology日志
        nimbus      启动nimbus
        rebalance   shell方式下修改topology运行参数比如worker个数 task个数等
        remoteconfvalue
        repl
        shell
        supervisor  启动supervisor
        ui              启动topology ui界面
        version

Help:
        help
        help <command>

 

 转载自http://www.cnblogs.com/quchunhui/p/5370191.html

--END--