理解Storm Topology的并发机制

缺乏、安全感 2022-04-14 03:24 210阅读 0赞

**运行时topology的组成：worker processes、executors(执行者线程)以及tasks**

Storm区分了以下三个主要的实体并用来在Storm集群上运行起一个topology:

1.Worker processes

2.Executors(线程)

3.Tasks

下面是这3者关系的一个简单示例:

![20151014122152718][]

一个worker process负责执行topology的部分子集。单个worker process归属于一个特定的topology并且可以为该topology的一个或多个组件(spouts或者bolts)运行一个或者多个executors。一个运行的topology由Storm集群中多台机器上运行着的多个processes(进程)组成。

一个executor其实就是由一个worker process生成的线程。executor将会为同一个组件（spout/bolt）运行多个tasks（任务）。

一个task负责进行数据处理---我们的代码实现的每个spout或者bolt会在集群上执行许多的任务。一个组件的tasks的数量在topology的整个生命周期中都是一样的。但是一个组件的executors的数量却会改变。这就意味着出出现这样一种情形:\#threads <= \#tasks。默认情况下，executor的数量与task的数量会设置成一样。例如:Storm中会每个executor运行一个task。

**topology的并发设置**

注意Storm中的术语"parallelism"被用来特指Storm中的parallelism hint，parallelism hint的意思是一个组件的初识executor的数量。这个文档中我们就用"parallelism"（并发）这个更宽泛的术语来描述：不仅仅限于executors的数量，还有worker processes的数量以及tasks的数量。在用"parallelism"来标识正常或者更特指的定义时，我们会特别提醒的。

接下来对多种配置选项的一个概览。我在Storm配置概览这篇文章中提到了Storm配置的优先级:defaults.yaml<storm.yaml<topology-specific configuration<internal component-specific configuration<external component-specific configuration。

**worker processes的数量**

**·**描述：为topology在集群机器上创建多少work processes。

·配置项:[TOPOLOGY\_WORKERS][TOPOLOGY_WORKERS]

·如何在代码中设置

*  [Config\#setNumWorkers][Config_setNumWorkers]

**executors（线程）的数量**

·描述：每个组件产生多少executors

·配置项：无（通过setSpout或者setBolt中的parallelism\_hint参数设置）

**task的数量**

·描述：为每个组件创建多少task

·配置项:[TOPOLOGY\_TASKS][TOPOLOGY_TASKS]

·如何在代码中设置(只是个例子):

·[ComponentConfigurationDeclarer\#setNumTasks()][ComponentConfigurationDeclarer_setNumTasks]

下面给出了这些配置的一个例子

1.  `topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2)`
2.  `.setNumTasks(4)`
3.  `.shuffleGrouping("blue-spout");`

在代码中我们配置了Storm中的bolt（GreenBolt）其中初始的executor数量为2以及相关taks数为4。Storm将会使每个executor运行两个task。如果没有显式的配置task数，Storm会每个executor运行一个task。

**一个运行中的topology示例**  
下面这个示例展示了一个简单的topology的运行。这个topology有3个组件组成:一个spout（BlueSpout)以及两个bolt(GreenBolt和YellowBolt)。由BlueSpout发送线性给GreenBolt，然后，GreenBolt发送消息给YellowBolt。

**![20151014141135164][]**

下面给出相应代码：

1.  `Config conf = new Config();`
2.  `conf.setNumWorkers(2); // use two worker processes`
3.   
4.  `topologyBuilder.setSpout("blue-spout", new BlueSpout(), 2); // set parallelism hint to 2`
5.   
6.  `topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2)`
7.  `.setNumTasks(4)`
8.  `.shuffleGrouping("blue-spout");`
9.   
10. `topologyBuilder.setBolt("yellow-bolt", new YellowBolt(), 6)`
11. `.shuffleGrouping("green-bolt");`
12.  
13. `StormSubmitter.submitTopology(`
14. `"mytopology",`
15. `conf,`
16. `topologyBuilder.createTopology()`
17. `);`

可以看到，BlueSpout与YellowBolt未设置task数，故默认task数与executor数相同。而GreenBolt设置了task数为4，故每个executor运行两个task。

Storm有额外的配置设置来控制topology的并发数:

·[TOPOLOGY\_MAX\_TASK\_PARALLELISM][TOPOLOGY_MAX_TASK_PARALLELISM]：这个设置对每个组件能生成的executor数设置了一个上限。这个典型的是在测试的时候，我们在本地模式下限制线程的数量。我们可以通过 [Config\#setMaxTaskParallelism()][Config_setMaxTaskParallelism]来进行设置。

**如何改变一个运行中的topology的并发量 **

Storm中的一个灵活的特征就是我们可以在不重启集群或者topology的情况下改变worker proceses以及executors的数量。这个过程叫做rebalancing(再均衡)

我们有两个选项来进行再均衡：

1.用Storm提供的web界面

2.通过命令行（CLI）工具

下面给出命令行工具的例子:

## Reconfigure the topology "mytopology" to use 5 worker processes,
    ## the spout "blue-spout" to use 3 executors and
    ## the bolt "yellow-bolt" to use 10 executors.
    
    $ storm rebalance mytopology -n 5 -e blue-spout=3 -e yellow-bolt=10

[20151014122152718]: /images/20220414/81920faab31d40eaa8a250fa415c96aa.png
[TOPOLOGY_WORKERS]: http://storm.apache.org/javadoc/apidocs/backtype/storm/Config.html#TOPOLOGY_WORKERS
[Config_setNumWorkers]: http://storm.apache.org/javadoc/apidocs/backtype/storm/Config.html
[TOPOLOGY_TASKS]: http://storm.apache.org/javadoc/apidocs/backtype/storm/Config.html#TOPOLOGY_TASKS
[ComponentConfigurationDeclarer_setNumTasks]: http://storm.apache.org/javadoc/apidocs/backtype/storm/topology/ComponentConfigurationDeclarer.html
[20151014141135164]: /images/20220414/52a52da898f748eb9abee390effc3457.png
[TOPOLOGY_MAX_TASK_PARALLELISM]: http://storm.apache.org/javadoc/apidocs/backtype/storm/Config.html#TOPOLOGY_MAX_TASK_PARALLELISM
[Config_setMaxTaskParallelism]: http://storm.apache.org/javadoc/apidocs/backtype/storm/Config.html#setMaxTaskParallelism%28int%29