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

野性酷女 2022-06-09 11:08 194阅读 0赞

[Storm和Zookeeper集群搭建及在java项目中的使用：][Storm_Zookeeper_java]  
[http://blog.csdn.net/xlgen157387/article/details/77540057][Storm_Zookeeper_java]

## 一、Storm架构简介 ##

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

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

![这里写图片描述][20170904192022799]

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

各个部分的含义如下：

（1）**Nodes（服务器）**：是指配置在一个Storm集群中的服务器，会执行topology的一部分运算。一个Storm集群可以包括一个或者多个工作node。

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

（3）**Executor(线程)**：是指一个worker的JVM进程中运行的Java线程。多个task可以指派给同一个executor来执行。除非是明确指定，Storm默认会给每一个executor分配一个task。

（4）**Task(bolt/spout实例)**：task是spout和bolt的实例，衙门的nextTuple()和execute()方法会被executors线程调用执行。

结构图如下：

![这里写图片描述][SouthEast]

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

以上一篇的代码为例，Topology的代码（代码A）如下：

![这里写图片描述][SouthEast 1]

可以看到Topology中有`RandomNameSpout`和两个Bolt：`UpperBolt`、`AppendBolt`，执行流程如下：

![这里写图片描述][SouthEast 2]

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

在这里，我们假设有一台服务器（Node节点），为topology分配了一个worker，并且每个executor执行一个task，那么上述代码（代码A）的执行流程如下图（**图A**）所示：

![这里写图片描述][SouthEast 3]

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

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

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

![这里写图片描述][SouthEast 4]

这样的话，整个topology就分配了2个worker而不是默认的1个。那么上图**图A**应该变成如下方式，如图（**图B**）：

![这里写图片描述][SouthEast 5]

## 四、Topology的并发机制–配置executor和task（1） ##

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

![这里写图片描述][SouthEast 6]

注意官方API对于`setSpout（）`方法的定义是这样的：

/**
         * 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.
         *
         * @param id the id of this component. This id is referenced by other components that want to consume this spout's outputs.
         * @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.
         * @param spout the spout
         * @throws IllegalArgumentException if {@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);
        }

![这里写图片描述][SouthEast 7]

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

那么上图**图A**应该变成如下方式，如图（**图C**）：

![这里写图片描述][SouthEast 8]

## 五、Topology的并发机制–配置executor和task（2） ##

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

![这里写图片描述][SouthEast 9]

那么上图**图A**应该变成如下方式，如图（**图D**）：

![这里写图片描述][SouthEast 10]

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

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

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

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

![这里写图片描述][SouthEast 11]

代码如下：

![这里写图片描述][SouthEast 12]

那么上图**图A**应该变成如下方式，如图（**图E**）：

![这里写图片描述][SouthEast 13]

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

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

参考文章：

1、[http://www.csdn.net/article/2015-03-09/2824135][http_www.csdn.net_article_2015-03-09_2824135]

[Storm_Zookeeper_java]: http://blog.csdn.net/xlgen157387/article/details/77540057
[20170904192022799]: /images/20220609/e85c06a28af94e489ba147f7819698da.png
[SouthEast]: /images/20220609/614dc1449878426da70168b49920ac7b.png
[SouthEast 1]: /images/20220609/c8c36b3d4bbf4bf9bfe4afca7d6c1053.png
[SouthEast 2]: /images/20220609/29e06f6abcb2480a9489fc3b494e3a55.png
[SouthEast 3]: /images/20220609/25debeec32914f2691b31b2651a2ae8a.png
[SouthEast 4]: /images/20220609/67a51cd05c17480bae81532c7a2ec2b6.png
[SouthEast 5]: /images/20220609/4d9e15a95061469ab8b574bedc27902a.png
[SouthEast 6]: /images/20220609/11c2fc450e2e4519b6891415daaf251b.png
[SouthEast 7]: /images/20220609/75484b57cf2f4c3a9cd5153a3e0b60d7.png
[SouthEast 8]: /images/20220609/91dcef1124044f6ca3bface76f96caa7.png
[SouthEast 9]: /images/20220609/0f9ceba9ea864477845e3e433763d4c2.png
[SouthEast 10]: /images/20220609/bdd9b9f28f2e41b19b26da4776cc9390.png
[SouthEast 11]: /images/20220609/86b8fc05322d4e4a958e5811dae5abae.png
[SouthEast 12]: /images/20220609/889c8d388d964df299a5a4b9d446e019.png
[SouthEast 13]: /images/20220609/f04f924c2fe64c8597c68107ea486c6d.png
[http_www.csdn.net_article_2015-03-09_2824135]: http://www.csdn.net/article/2015-03-09/2824135