57.storm拓扑结构调整

首页 > 代码库 > 57.storm拓扑结构调整

2024-09-23 18:34:06 213人阅读

几个概念

Topology（拓扑）：Spout、Bolt组成的一个完整的流程结构；

Stream Grouping：流分组、数据的分发方式；

Spout：直译水龙头，也就是消息源的意思；

Bolt：螺栓、处理器。很形象，水从上面的那个“水龙头”流出来，经过第一个螺栓，经过第二个螺栓，经过第三第四个螺栓...

Worker：工作进程

Executor：执行器、task的线程；

Task：具体执行的任务；

Configuration：配置。

实际操作

回顾

在上上节（55节），本地模式跑起来的，只有一个JVM（虽然分配了两个，cfg.setNumWorkers(2); 但是在本地跑的时候只可能启动一个JVM），那么拓扑的执行情况应该是下面这样的，一个worker下有几个Executor，每个Executor分别对应一个Spout或者Bolt（图片中的bolt和Spout名称和代码不一致，请对号入座）：

技术分享

结构调整

如果我们把带码稍作改动（只需要改Topology）

 1 import backtype.storm.Config;
 2 import backtype.storm.LocalCluster;
 3 import backtype.storm.StormSubmitter;
 4 import backtype.storm.topology.TopologyBuilder;
 5 import bhz.bolt.PrintBolt;
 6 import bhz.bolt.WriteBolt;
 7 import bhz.spout.PWSpout;
 8 
 9 public class PWTopology2 {
10 
11     public static void main(String[] args) throws Exception {
12         
13         Config cfg = new Config();
14         cfg.setNumWorkers(2);//设置使用俩个工作进程
15         cfg.setDebug(false);
16         TopologyBuilder builder = new TopologyBuilder();
17         //设置sqout的并行度和任务数（产生2个执行器和俩个任务）
18         builder.setSpout("spout", new PWSpout(), 2);//.setNumTasks(2);
19         //设置bolt的并行度和任务数:（产生2个执行器和4个任务）
20         builder.setBolt("print-bolt", new PrintBolt(), 2).shuffleGrouping("spout").setNumTasks(4);
21         //设置bolt的并行度和任务数:（产生6个执行器和6个任务）
22         builder.setBolt("write-bolt", new WriteBolt(), 6).shuffleGrouping("print-bolt");
23         
24         
25         //1 本地模式
26 //        LocalCluster cluster = new LocalCluster();
27 //        cluster.submitTopology("top2", cfg, builder.createTopology());
28 //        Thread.sleep(10000);
29 //        cluster.killTopology("top2");
30 //        cluster.shutdown();
31         
32         //2 集群模式
33         StormSubmitter.submitTopology("top2", cfg, builder.createTopology());
34         
35     }
36 }