第3节 storm高级应用：1、上次课程回顾，今日课程大纲，storm下载地址、运行过程等...

偏执的太偏执、 2021-11-17 05:08 99阅读 0赞

上次课程内容回顾：

ConcurrentHashMap是线程安全的，为什么多线程的时候还不好使，为什么还要加static关键字

1、storm的基本介绍：strom是twitter公司开源提供给apache的一个实时处理的框架  
2、storm的架构模型：主从架构：  
 nimbus：主节点  接收客户端提交的任务，以及任务的分发  
 supervisor：从节点，主要用于处理nimbus分配的任务  
3、storm的安装 yaml的配置文件比较严格  
4、strom的UI管理界面：记得更改ui的端口，默认8088已经被占用了  
5、strom的编程模型：  
 spout：接收数据源  
 bolt：处理我们的数据的组件  
6、storm的入门程序：wordCount  
7、storm的并行度。调整进程和线程的数量  
8、storm的分发策略  
9、storm与kafka的集成   KafkaSpout去消费kafka当中的数据就行了  
10、实时看板综合案例

==========================================

![1325651-20190728232424400-829059369.png][]

==========================================

# Storm实时处理第二天 #

# 1、Storm源码下载及目录熟悉 #

### 1.1、在Storm官方网站上寻找源码地址 ###

[http://storm.apache.org/downloads.html][http_storm.apache.org_downloads.html]

![1325651-20190728230943601-1281010827.png][]

### 1.2 点击文字标签进入github ###

点击Apache/storm文字标签，进入github

[https://github.com/apache/storm][https_github.com_apache_storm]

### 1.3 拷贝storm源码地址 ###

在网页右侧，拷贝storm源码地址

![1325651-20190728231121720-1147071896.png][]

### 1.4 使用Subversion客户端下载 ###

![1325651-20190728231146921-1875538265.png][]

[https://github.com/apache/storm/tags/v0.9.5][https_github.com_apache_storm_tags_v0.9.5]

### 1.5 Storm源码目录分析 ###

![1325651-20190728231329743-785109186.png][]

扩展包中的三个项目，使storm能与hbase、hdfs、kafka交互

![1325651-20190728231345550-1234824494.png][]

### 1.6 Storm源码编译 ###

![1325651-20190728231432092-1898154372.png][]

# 2、Storm原理 #

### 2.1  Storm 任务提交的过程 ###

![1325651-20190728231753441-793984876.png][]

<table> 
 <tbody> 
  <tr> 
   <td> <p>TopologyMetricsRunnable.TaskStartEvent[oldAssignment=&lt;null&gt;,newAssignment=Assignment[masterCodeDir=C:\Users\MAOXIA~1\AppData\Local\Temp\\e73862a8-f7e7-41f3-883d-af494618bc9f\nimbus\stormdist\double11-1-1458909887,nodeHost={61ce10a7-1e78-4c47-9fb3-c21f43a331ba=192.168.1.106},taskStartTimeSecs={1=1458909910, 2=1458909910, 3=1458909910, 4=1458909910, 5=1458909910, 6=1458909910, 7=1458909910, 8=1458909910},workers=[ResourceWorkerSlot[hostname=192.168.1.106,memSize=0,cpu=0,tasks=[1, 2, 3, 4, 5, 6, 7, 8],jvm=&lt;null&gt;,nodeId=61ce10a7-1e78-4c47-9fb3-c21f43a331ba,port=6900]],timeStamp=1458909910633,type=Assign],task2Component=&lt;null&gt;,clusterName=&lt;null&gt;,topologyId=double11-1-1458909887,timestamp=0]</p> </td> 
  </tr> 
 </tbody> 
</table>

![1325651-20190728232014524-1525789169.png][]

<table> 
 <tbody> 
  <tr> 
   <td> <p><span style="font-size:14px;">TopologyMetricsRunnable.TaskStartEvent[oldAssignment=&lt;null&gt;,newAssignment=Assignment[masterCodeDir=C:\Users\MAOXIA~1\AppData\Local\Temp\\e73862a8-f7e7-41f3-883d-af494618bc9f\nimbus\stormdist\double11-1-1458909887,nodeHost={61ce10a7-1e78-4c47-9fb3-c21f43a331ba=192.168.1.106},taskStartTimeSecs={1=1458909910, 2=1458909910, 3=1458909910, 4=1458909910, 5=1458909910, 6=1458909910, 7=1458909910, 8=1458909910},workers=[ResourceWorkerSlot[hostname=192.168.1.106,memSize=0,cpu=0,tasks=[1, 2, 3, 4, 5, 6, 7, 8],jvm=&lt;null&gt;,nodeId=61ce10a7-1e78-4c47-9fb3-c21f43a331ba,port=6900]],timeStamp=1458909910633,type=Assign],task2Component=&lt;null&gt;,clusterName=&lt;null&gt;,topologyId=double11-1-1458909887,timestamp=0]</span></p> </td> 
  </tr> 
 </tbody> 
</table>

![1325651-20190728232038683-7664557.png][]

### 2.2  Storm组件本地目录树 ###

![1325651-20190728232600697-1060157242.png][]

### 2.3  Storm zookeeper目录树 ###

![1325651-20190728232902932-574817337.png][]

### 2.4  Storm启动流程分析 ###

\------------程序员client------------------

1、客户端运行storm nimbus时，会调用storm的python脚本，该脚本中为每个命令编写一个方法，每个方法都可以生成一条相应的java命令。

命令格式如下：java -server xxxx.ClassName -args

nimbus---> Running: /export/servers/jdk/bin/java -server  backtype.storm.daemon.nimbus

supervisor---> Running: /export/servers/jdk/bin/java -server  backtype.storm.daemon.supervisor

\--------------nimbus---------------------

2、nibums启动之后，接受客户端提交任务

命令格式：storm jar xxx.jar   xxx驱动类  参数

Running: /export/servers/jdk/bin/java -client -Dstorm.jar=/export/servers/storm/examples/storm-starter/storm-starter-topologies-0.9.6.jar storm.starter.WordCountTopology wordcount-28

该命令会执行 storm-starter-topologies-0.9.6.jar 中的storm-starter-topologies-0.9.6.jar的main方法，main方法中会执行以下代码：

StormSubmitter.submitTopology("mywordcount",config,topologyBuilder.createTopology());

topologyBuilder.createTopology()，会将程序猿编写的spout对象和bolt对象进行序列化。

会将用户的jar上传到 nimbus物理节点的 /export/data/storm/workdir/nimbus/inbox目录下。并且改名，改名的规则是添加了一个UUID字符串。

在nimbus物理节点的 /export/data/storm/workdir/nimbus/stormdist目录下。有当前正在运行的topology的jar包和配置文件，序列化对象文件。

3、nimbus接受到任务之后，会将任务进行分配，分配会产生一个assignment对象，该对象会保存到zk中，目录是/storm/assignments ，该目录只保存正在运行的topology任务。

\--------supervisor------------------

4、supervisor通过watch机制，感知到nimbus在zk上的任务分配信息，从zk上拉取任务信息，分辨出属于自己任务。

ResourceWorkerSlot\[hostname=192.168.1.106,memSize=0,cpu=0,tasks=\[1, 2, 3, 4, 5, 6, 7, 8\],jvm=<null>,nodeId=61ce10a7-1e78-4c47-9fb3-c21f43a331ba,port=6900\]

5、supervisor 根据自己的任务信息，启动自己的worker，并分配一个端口。

'/export/servers/jdk/bin/java' '-server' '-Xmx768m' export/data/storm/workdir/supervisor/stormdist/wordcount1-3-1461683066/stormjar.jar' 'backtype.storm.daemon.worker' 'wordcount1-3-1461683066' 'a69bb8fc-e08e-4d55-b51f-e539b066f90b' '6701' '9fac2805-7d2b-4e40-aabc-1c85c9856d64'

\---------worker----------------------

6、worker启动之后，连接zk，拉取任务

ResourceWorkerSlot\[hostname=192.168.1.106,memSize=0,cpu=0,tasks=\[1, 2, 3, 4, 5, 6, 7, 8\],jvm=<null>,nodeId=61ce10a7-1e78-4c47-9fb3-c21f43a331ba,port=6900\]

假设任务信息：

1--->spout---type:spout

2--->bolt ---type:bolt

3--->acker---type:bolt

得到对象有几种方式？ new ClassName 创建对象、class.forName 反射对象、clone 克隆对象、序列化反序列化

worker通过反序列化，得到程序员自己定义的spout和bolt对象。

7、worker根据任务类型，分别执行spout任务或者bolt任务。

spout的声明周期是：open、nextTuple、outPutFiled

bolt的生命周期是：prepare、execute(tuple)、outPutFiled

### 2.5  启动流程代码说明 ###

jstorm supervisor如何启动worker，worker如何启动task

1、下载Jstorm源码，在源码包下找到 daemon包，在这个包下有三个子包，分别是nimbus，supervisor，worker。

2、通过架构图，我们已知nimbus分配任务，并将任务信息写入到zk上，supervisor读取zk上的任务后启动自己的worker。所以我们分析supervisor如何启动worker，worker如何启动task。

3、supervisor如何启动worker。打开 com.alibaba.jstorm.daemon.supervisor.Supervisor 发现supervisor有几个方法，方法中有个mkSupervisor方法。

4、进去Supervisor中的mkSupervisor方法，在第144行有以下的代码，改代码创建了SyncSupervisorEvent 对象。

SyncSupervisorEvent syncSupervisorEvent =

new SyncSupervisorEvent(supervisorId, conf, syncSupEventManager, stormClusterState, localState, syncProcessEvent, hb);

5、SyncSupervisorEvent对象实现了RunnableCallback接口，该接口有个run方法会被定时执行。在run方法的191行，有代码如下，主要是要supervisor获取到任务信息，要开始准备启动worker了。

syncProcesses.run(zkAssignment, downloadFailedTopologyIds);

6、syncProcesses是com.alibaba.jstorm.daemon.supervisor.SyncProcessEvent的

引用变量，该类中有个自定义的run方法中有段代码如下，调用的startNewWorkers方法

startNewWorkers(keepPorts, localAssignments, downloadFailedTopologyIds);

7、SyncProcessEvent的startNewWorkers方法有代码片段如下，主要是根据集群模式启动不同模式下的worker。我们跟踪分布式集群模式下的worker启动。

for (Entry<Integer, LocalAssignment> entry : newWorkers.entrySet()) \{

if (clusterMode.equals(“distributed”)) \{

launchWorker(conf, sharedContext, assignment.getTopologyId(), supervisorId, port, workerId, assignment);

\} else if (clusterMode.equals(“local”)) \{

launchWorker(conf, sharedContext, assignment.getTopologyId(), supervisorId, port, workerId, workerThreadPids);

8、在分布式模式下worker启动最终会调用一个类似于java -server xxx.worker 启动worker。由于第7步中，有个for循环，该for循环会迭代出属于当前supervisor的所有worker任务并启动。

JStormUtils.launchProcess(cmd, environment, true);

9、java -server xxx.worker，命令执行之后，会执行Worker的mian方法。worker的main方法有代码如下，其实调用了worker自己内部的静态方法，叫做mk\_worker方法。

WorkerShutdown sd = mk\_worker(conf, null, topology\_id, supervisor\_id, Integer.parseInt(port\_str), worker\_id, jar\_path);

sd.join();

10、mk\_worker静态方法，会执行以下代码，创建一个worker的实例，并立即执行execute方法。

Worker w = new Worker(conf, context, topology\_id, supervisor\_id, port, worker\_id, jar\_path);

return w.execute();

11、execute方法会执行以下代码创建一个RefreshConnections 的实例。

RefreshConnections refreshConn = makeRefreshConnections();

12、makeRefreshConnections 方法会执行以下代码创建一个RefreshConnections 实例。

RefreshConnections refresh\_connections = new RefreshConnections(workerData);

13、RefreshConnections 是继承了 RunnableCallback，该实例的会有一个run方法会被定时执行。run方法中有以下代码，其中createTasks(addedTasks)方法用来创建Task任务。

shutdownTasks(removedTasks);

createTasks(addedTasks);

updateTasks(updatedTasks);

14、createTasks方法有代码如下，循环启动属于该worker的Task任务，启动Task任务主要调用Task.mk\_task(workerData, taskId);

for (Integer taskId : tasks) \{

try \{

TaskShutdownDameon shutdown = Task.mk\_task(workerData, taskId);

workerData.addShutdownTask(shutdown);

\} catch (Exception e) \{

LOG.error(“Failed to create task-” + taskId, e);

throw new RuntimeException(e);

15、Task.mk\_task(workerData, taskId)方法实现如下，创建一个Task对象并立即调用execute方法。

Task t = new Task(workerData, taskId);

return t.execute();

16、execute方法实现如下,用来初始化一个Executor，我们知道在默认情况下一个task等于一个executor。

RunnableCallback baseExecutor = prepareExecutor();

17、进入prepareExecutor()方法，代码如下，发现代码调用了mkExecutor方法。

final BaseExecutors baseExecutor = mkExecutor();

18、mkExecutor方法，代码如下，如果当前taskObj是Bolt就创建Bolt的executor，如果当前taskObj是Spout就创建相应的Spout executor。

public BaseExecutors mkExecutor() \{

BaseExecutors baseExecutor = null;

if (taskObj instanceof IBolt) \{

baseExecutor = new BoltExecutors(this);

\} else if (taskObj instanceof ISpout) \{

if (isSingleThread(stormConf) == true) \{

baseExecutor = new SingleThreadSpoutExecutors(this);

\} else \{

baseExecutor = new MultipleThreadSpoutExecutors(this);

return baseExecutor;

19、创建完了executor，现在有两条线，分别是bolt executor和spout executor。以

bolt executor 为例，这个executor会实现Disruptor的EventHandler接口。 接口onevent方法需要实现，实现代码中会调用processTupleEvent()方法。下面节选onevent中的部分代码。

if (event instanceof Tuple) \{

processControlEvent();

processTupleEvent((Tuple) event);

\} else if (event instanceof BatchTuple) \{

for (Tuple tuple : ((BatchTuple) event).getTuples()) \{

processControlEvent();

processTupleEvent((Tuple) tuple);

20、进入processTupleEvent方法，发现有代码如下，其实最终是调用了bolt.execute()方法。

private void processTupleEvent(Tuple tuple) \{

try \{

if (xxx) \{

backpressureTrigger.handle(tuple);

\} else \{

bolt.execute(tuple);

\} catch (Throwable e) \{

error = e;

LOG.error(“bolt execute error “, e);

report\_error.report(e);

转载于:https://www.cnblogs.com/mediocreWorld/p/11261410.html

[1325651-20190728232424400-829059369.png]: /images/20211116/2ef8a4ed004a4db097070520d906a42f.png
[http_storm.apache.org_downloads.html]: http://storm.apache.org/downloads.html
[1325651-20190728230943601-1281010827.png]: /images/20211116/79b90ec758464b4794819baf4cab79f8.png
[https_github.com_apache_storm]: https://github.com/apache/storm
[1325651-20190728231121720-1147071896.png]: /images/20211116/50c4b639875d4c2fbdc918f5ddc4f862.png
[1325651-20190728231146921-1875538265.png]: /images/20211116/345c7f82e4504df68ad463d86033b611.png
[https_github.com_apache_storm_tags_v0.9.5]: https://github.com/apache/storm/tags/v0.9.5
[1325651-20190728231329743-785109186.png]: /images/20211116/7d20626a19e942ff986bac3a6f74f136.png
[1325651-20190728231345550-1234824494.png]: /images/20211116/f2c09e43d2e14377b8a5bf832a0dd978.png
[1325651-20190728231432092-1898154372.png]: /images/20211116/f81173915d984e4aaf2af08a8f96bc77.png
[1325651-20190728231753441-793984876.png]: /images/20211116/c960a214d2b84cd0ae7be18f81266257.png
[1325651-20190728232014524-1525789169.png]: /images/20211116/51ee72d036a54c339cce3940808fe27a.png
[1325651-20190728232038683-7664557.png]: /images/20211116/257ddf44a4604b9f9787d0fc7ed90b6c.png
[1325651-20190728232600697-1060157242.png]: /images/20211116/92f01004be1b464dbd8ce4651bdb31fd.png
[1325651-20190728232902932-574817337.png]: /images/20211116/b4e8d8f266e4481bbdbee556e393720a.png