图解Storm并发机制及其执行流程

上一篇&＃xff1a;

Storm和Zookeeper集群搭建及在java项目中的使用&＃xff1a;
http://blog.csdn.net/xlgen157387/article/details/77540057

一、Storm架构简介

在上一篇&＃xff0c;我们对Storm集群进行了搭建&＃xff0c;并使用Java完成了代码的演示&＃xff0c;我们知道在Storm中&＃xff0c;先要设计一个用于实时计算的图状结构&＃xff0c;我们称之为拓扑&＃xff08;topology&＃xff09;。这个拓扑将会被提交给集群&＃xff0c;由集群中的主控节点&＃xff08;master node&＃xff09;分发代码&＃xff0c;将任务分配给工作节点&＃xff08;worker node&＃xff09;执行。

一个拓扑中包括spout和bolt两种角色&＃xff0c;其中spout发送消息&＃xff0c;负责将数据流以tuple元组的形式发送出去&＃xff1b;而bolt则负责转换这些数据流&＃xff0c;在bolt中可以完成计算、过滤等操作&＃xff0c;bolt自身也可以随机将数据发送给其他bolt。由spout发射出的tuple是不可变数组&＃xff0c;对应着固定的键值对。

在Storm中&＃xff0c;一个task可以简单的理解为在集群某节点上运行的一个spout或者bolt实例。在集群运行运行中&＃xff0c;topology主要有四个组成部分&＃xff1a;他们从低到高分别是&＃xff1a;task(bolt/spout实例)、Executor(线程)、Workers(JVM虚拟机)、Nodes(服务器)

各个部分的含义如下&＃xff1a;

&＃xff08;1&＃xff09;Nodes&＃xff08;服务器&＃xff09;&＃xff1a;是指配置在一个Storm集群中的服务器&＃xff0c;会执行topology的一部分运算。一个Storm集群可以包括一个或者多个工作node。

&＃xff08;2&＃xff09;Workers&＃xff08;JVM虚拟机&＃xff09;&＃xff1a;是指一个node节点服务器上相互独立运行的JVM进程。每一个node可以配置运行一个或者多个worker。一个topology会分配到一个或者多个worker上运行。

&＃xff08;3&＃xff09;Executor(线程)&＃xff1a;是指一个worker的JVM进程中运行的Java线程。多个task可以指派给同一个executor来执行。除非是明确指定&＃xff0c;Storm默认会给每一个executor分配一个task。

&＃xff08;4&＃xff09;Task(bolt/spout实例)&＃xff1a;task是spout和bolt的实例&＃xff0c;衙门的nextTuple()和execute()方法会被executors线程调用执行。

结构图如下&＃xff1a;

二、Topology的并发机制–默认配置

以上一篇的代码为例&＃xff0c;Topology的代码&＃xff08;代码A&＃xff09;如下&＃xff1a;

可以看到Topology中有RandomNameSpout和两个Bolt&＃xff1a;UpperBolt、AppendBolt&＃xff0c;执行流程如下&＃xff1a;

上边的代码在默认的情况下&＃xff0c;我们没有使用Storm中并发机制提供的API&＃xff0c;全部都是默认的&＃xff0c;在大多数情况下&＃xff0c;除非明确指定&＃xff0c;Storm的默认并发设置是1。

在这里&＃xff0c;我们假设有一台服务器&＃xff08;Node节点&＃xff09;&＃xff0c;为topology分配了一个worker&＃xff0c;并且每个executor执行一个task&＃xff0c;那么上述代码&＃xff08;代码A&＃xff09;的执行流程如下图&＃xff08;图A&＃xff09;所示&＃xff1a;

由上图&＃xff0c;我们可以看出&＃xff0c;唯一的并发机制出现在线程级。每个任务Task在同一个JVM的不同线程中执行。

三、Topology的并发机制–给Topology增加Worker

增加额外的worker是增加topology计算能力的简单方式&＃xff0c;Storm提供了简单的配置使我们增加worker的方式变得很容易&＃xff0c;只需修改如下代码即可&＃xff0c;其它代码不变&＃xff1a;

这样的话&＃xff0c;整个topology就分配了2个worker而不是默认的1个。那么上图图A应该变成如下方式&＃xff0c;如图&＃xff08;图B&＃xff09;&＃xff1a;

四、Topology的并发机制–配置executor和task&＃xff08;1&＃xff09;

Storm并发机制API允许设定每个task对应的executor个数和每个executor可执行的task个数。在定义流分组的时候&＃xff0c;也可以设置每一个组件指派的executor的数量。例如&＃xff1a;我们修改RandomNameSpout并发为两个task&＃xff0c;每个task指派各自的executor线程&＃xff0c;还是只使用1个worker代码修改如下&＃xff1a;

注意官方API对于setSpout&＃xff08;&＃xff09;方法的定义是这样的&＃xff1a;

/**
* Define a new spout in this topology with the specified parallelism. If the spout declares
* itself as non-distributed, the parallelism_hint will be ignored and only one task
* will be allocated to this component.
*
* &＃64;param id the id of this component. This id is referenced by other components that want to consume this spout&＃39;s outputs.
* &＃64;param parallelism_hint the number of tasks that should be assigned to execute this spout. Each task will run on a thread in a process somewhere around the cluster.
* &＃64;param spout the spout
* &＃64;throws IllegalArgumentException if {&＃64;code parallelism_hint} is not positive
*/
public SpoutDeclarer setSpout(String id, IRichSpout spout, Number parallelism_hint) throws IllegalArgumentException {
validateUnusedId(id);
initCommon(id, spout, parallelism_hint);
_spouts.put(id, spout);
return new SpoutGetter(id);
}

对于第三个参数&＃xff0c;意思是&＃xff1a;设置Spout的并发为两个task&＃xff0c;每个task指派给各自的executor线程&＃xff0c;由于默认情况下&＃xff0c;每一个线程executor执行一个task&＃xff0c;所以我们可以理解为&＃xff0c;分配了两个线程executor&＃xff0c;每一个executor线程执行一个任务task。

那么上图图A应该变成如下方式&＃xff0c;如图&＃xff08;图C&＃xff09;&＃xff1a;

五、Topology的并发机制–配置executor和task&＃xff08;2&＃xff09;

上述&＃xff0c;增加了Spout的线程数&＃xff0c;在默认情况下每一个线程executor是处理一个task&＃xff0c;那么&＃xff0c;我们接下来设置多个executor和多个task&＃xff0c;完整代码如下&＃xff1a;

那么上图图A应该变成如下方式&＃xff0c;如图&＃xff08;图D&＃xff09;&＃xff1a;

有上述代码&＃xff0c;可知设置了2个worker&＃xff0c;因此每一个worker平均分摊执行相应的task&＃xff0c;最后的结果就如上图所示。

值得注意的是&＃xff1a;如果只在一台Node服务器上增加worker的数量&＃xff0c;例如&＃xff1a;topology执行在本地模式的时候&＃xff0c;并不会显著地提升系统的性能&＃xff0c;也是会出现瓶颈的&＃xff0c;这是因为topology在本地模式下是在同一个JVM进程中执行的&＃xff0c;必然会有相关资源的竞争等&＃xff0c;所以只有增加task和executor的并发配置才会生效。

六、Topology的并发机制的特殊情况

实际开发中可能存在这种情况&＃xff0c;例如&＃xff1a;我们在上述的基础上在增加一个类似于汇总的Bolt进行统计字符串的多少&＃xff0c;那我&＃xff0c;我们只可能对这个Bolt设置为单线程、单任务的方式进行统计&＃xff0c;而不可能在使用多线程或多任务的方式进行统计&＃xff0c;原因很简单&＃xff0c;执行示意图如下&＃xff1a;

代码如下&＃xff1a;

那么上图图A应该变成如下方式&＃xff0c;如图&＃xff08;图E&＃xff09;&＃xff1a;

由此&＃xff0c;我们可以知道&＃xff0c;Storm提供了简单的API接口让我们能够很方便的进行并发控制&＃xff0c;但是我们也要根据具体的业务设置合理的executor和task等数目&＃xff0c;以免发生错误的结果。

参考文章&＃xff1a;

1、http://www.csdn.net/article/2015-03-09/2824135