Topology的并行度设置

一个运行的topology由3类物理实体构成

• Worker进程
• Executor线程
• Task实例

当运行一个topology的时候，首先会在storm集群中启多个worker进程，每个worker进程中再起若干的executor线程，每个executor线程中运行一个或多个task的实例，每个executor中的task都属于同一个spout或者bolt。默认每个executor中运行一个task实例。

每个spout或者bolt都会被创建多个task实例执行具体的业务逻辑，注意一个executor中如果有多个task，它们之间是串行的。

样例

Config conf = new Config();
conf.setNumWorkers(2); // 设置两个worker进程

topologyBuilder.setSpout("blue-spout", new BlueSpout(), 2); // 设置并行度为2，这里指的就是executor的数量

topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2)
               .setNumTasks(4) //设置这个bolt的task总数为4
               .shuffleGrouping("blue-spout");

topologyBuilder.setBolt("yellow-bolt", new YellowBolt(), 6)
               .shuffleGrouping("green-bolt");

StormSubmitter.submitTopology(
        "mytopology",
        conf,
        topologyBuilder.createTopology()
    );

通过上面的代码片度，可以计算并行度

说明

通过上面的解释对worker和executor的概念很好理解，这里着重说明一下task。为了提高并行度，对于运行中的topology官方提供了动态更新worker和executor数量的方法，但是并没有说tasks的数量在提高并行度中起到的作用，下面是引用官方对task的解释

A task performs the actual data processing — each spout or bolt that you implement in your code executes as many tasks across the cluster. ***The number of tasks for a component is always the same throughout the lifetime of a topology, but the number of executors (threads) for a component can change over time. This means that the following condition holds true: #threads ≤ #tasks. ***By default, the number of tasks is set to be the same as the number of executors, i.e. Storm will run one task per thread.

这里涉及到几个关键点：

• task的数量在整个topology的生命周期中是不会改变的
• executor的数量一定小于等于task的数量，大于task的数量没有任何意义

初始化配置的时候如果task的数量大于executor的数量，例如上例中的green-bolt。那么一个executor中只能顺序执行其中的task，通过rebalance命令重置executor达到和task的数量一致，那么每个executor执行一个task，这样所有的task就都可以并行了，提高了topology的并行度。

但是像上例中的blue-spout和yellow-bolt，初始化的时候没有指定task的数量，那么默认task的数量和executor数量是一样的，即使通过rebalance命令提高executor的数量，对于并行度也是没有任何影响的，threads的数量已经等于task了。

所以设置task的数量是为了以后提高并行度做准备，一旦topology运行之后task的数量就无法改变了，如果要通过rebalance命令提高并行度，初始设置的时候task的数量要大于executor的数量。