storm java 编程思路

整体思路

storm编程和hadoop的mapreduce的编程很相似，hadoop的mapreduce须要本身实现map函数，reduce函数，还有一个主类驱动；storm须要本身实现spout，bolt和一个主函数。storm编程为如下三步：

建立一个Spout读取数据
建立bolt处理数据
建立一个主类，在主类中建立拓扑和一个集群对象，将拓扑提交到集群

Topology运行方式
Topology的运行能够分为本地模式和分布式模式，模式的设置能够在配置文件中设定，也能够在代码中设置。本地模式其实什么都不须要安装，有storm jar包就够了 

(1)本地运行的提交方式：
LocalCluster cluster = new LocalCluster();
cluster.submitTopology(topologyName, conf, topology);
cluster.killTopology(topologyName);
cluster.shutdown();
 
(2)分布式提交方式：
StormSubmitter.submitTopology(topologyName, topologyConfig, builder.createTopology());
须要注意的是，在Storm代码编写完成以后，须要打包成jar包放到Nimbus中运行，打包的时候，不须要把依赖的jar都打进去，不然若是把依赖的storm.jar包打进去的话，运行时会出现重复的配置文件错误致使Topology没法运行。由于Topology运行以前，会加载本地的storm.yaml配置文件。
在Nimbus运行的命令以下：
storm jar StormTopology.jar maincalss args
Topology运行流程
　　有几点须要说明的地方：
(1)Storm提交后，会把代码首先存放到Nimbus节点的inbox目录下，以后，会把当前Storm运行的配置生成一个stormconf.ser文件放到Nimbus节点的stormdist目录中，在此目录中同时还有序列化以后的Topology代码文件；
(2)在设定Topology所关联的Spouts和Bolts时，能够同时设置当前Spout和Bolt的executor数目和task数目，默认状况下，一个Topology的task的总和是和executor的总和一致的。以后，系统根据worker的数目，尽可能平均的分配这些task的执行。worker在哪一个supervisor节点上运行是由storm自己决定的；
(3)任务分配好以后，Nimbes节点会将任务的信息提交到zookeeper集群，同时在zookeeper集群中会有workerbeats节点，这里存储了当前Topology的全部worker进程的心跳信息；
(4)Supervisor节点会不断的轮询zookeeper集群，在zookeeper的assignments节点中保存了全部Topology的任务分配信息、代码存储目录、任务之间的关联关系等，Supervisor经过轮询此节点的内容，来领取本身的任务，启动worker进程运行；
(5)一个Topology运行以后，就会不断的经过Spouts来发送Stream流，经过Bolts来不断的处理接收到的Stream流，Stream流是无界的。
最后一步会不间断的执行，除非手动结束Topology。
Topology方法调用流程
Topology中的Stream处理时的方法调用过程以下：

 有几点须要说明的地方：
   (1)每一个组件(Spout或者Bolt)的构造方法和declareOutputFields方法都只被调用一次。
   (2)open方法、prepare方法的调用是屡次的。入口函数中设定的setSpout或者setBolt里的并行度参数指的是executor的数目，是负责运行组件中的task的线程         的数目，此数目是多少，上述的两个方法就会被调用多少次，在每一个executor运行的时候调用一次。至关于一个线程的构造方法。
   (3)nextTuple方法、execute方法是一直被运行的，nextTuple方法不断的发射Tuple，Bolt的execute不断的接收Tuple进行处理。只有这样不断地运行，才会产生无界的Tuple流，体现实时性。至关于线程的run方法。
   (4)在提交了一个topology以后，Storm就会建立spout/bolt实例并进行序列化。以后，将序列化的component发送给全部的任务所在的机器(即Supervisor节点)，在每个任务上反序列化component。
   (5)Spout和Bolt之间、Bolt和Bolt之间的通讯，是经过zeroMQ的消息队列实现的。
   (6)上图没有列出ack方法和fail方法，在一个Tuple被成功处理以后，须要调用ack方法来标记成功，不然调用fail方法标记失败，从新处理这个Tuple。
Topology并行度
    在Topology的执行单元里，有几个和并行度相关的概念。
(1)worker:每一个worker都属于一个特定的Topology，每一个Supervisor节点的worker能够有多个，每一个worker使用一个单独的端口，它对Topology中的每一个component运行一个或者多个executor线程来提供task的运行服务。
(2)executor：executor是产生于worker进程内部的线程，会执行同一个component的一个或者多个task。
(3)task:实际的数据处理由task完成，在Topology的生命周期中，每一个组件的task数目是不会发生变化的，而executor的数目却不必定。executor数目小于等于task的数目，默认状况下，两者是相等的。
    在运行一个Topology时，能够根据具体的状况来设置不一样数量的worker、task、executor，而设置的位置也能够在多个地方。
(1)worker设置：
(1.1)能够经过设置yaml中的topology.workers属性
(1.2)在代码中经过Config的setNumWorkers方法设定
(2)executor设置：
    经过在Topology的入口类中setBolt、setSpout方法的最后一个参数指定，不指定的话，默认为1；
(3)task设置：
    (3.1) 默认状况下，和executor数目一致；
    (3.2)在代码中经过TopologyBuilder的setNumTasks方法设定具体某个组件的task数目；
终止Topology
    经过在Nimbus节点利用以下命令来终止一个Topology的运行：
storm kill topologyName
    kill以后，能够经过UI界面查看topology状态，会首先变成KILLED状态，在清理完本地目录和zookeeper集群中的和当前Topology相关的信息以后，此Topology就会完全消失了。
Topology跟踪

    Topology提交后，能够在Nimbus节点的web界面查看，默认的地址是http://NimbusIp:8080。