使用Storm实现WordSum

文章目录

一、Spout类
二、Bolt类
三、源码实现
- 3.1 MySpout 类
3.2 MyBolt 类
3.3 测试类

一、Spout类

Spout类extends BaseRichSpout，BaseRichSpout extends BaseComponent implements IRichSpout，IRichSpout extends ISpout；分析ISpout几个方法：

1、void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
任务调用的时候，在一个worker上初始化；提供了集群拓扑作业的配置信息、当前作业的任务信息、collector用来发送封装的tuples单元
2、void nextTuple();
strom要求spout发送数据给output collector，非阻塞式方法，如果没有数据发送，该方法就会return；

二、Bolt类

Bolt类extends BaseRichBolt,BaseRichBolt实现IRichBolt，IRichBolt继承IBolt，接下来分析IBolt几个方法：

1、 void prepare(Map stormConf, TopologyContext context, OutputCollector collector);
根据注释，当任务来了的时候，这个会在集群中的某个worker节点被初始化，他提供了Bolt的运行环境。3个参数：stormConf给当前Bolt准备的配置对象；context可以获取任务的位置信息，包括任务id和组件id、输入输出流的信息；collector发送数据

2、 void execute(Tuple input);
处理一个单一的tuple输入流，元组对象包含了metadata元数据信息（封装了发送的数据的来自于哪个组件、哪个流、哪个任务），value值可以被获取到Tuple对象的getValue方法

当然，它们都有一个共同的方法-declareOutputFields，负责给所有的流声明了output输出策略。

三、源码实现

3.1 MySpout 类

在这个类中，我们首先要在open方法中初始化，然后在nextTuple方法中，不停的采集数据、向后发送数据。在调用collector的emit方法向后发射数据的时候，要对后面的Bolt声明发送数据的字段名称。类似于Android中使用Intent、SP传值时定义的类型、名称。

public class MySpout extends BaseRichSpout{Map map;TopologyContext context;SpoutOutputCollector collector;int i=0;/*** 配置初始化spout类,提升作用域* */@Overridepublic void open(Map map, TopologyContext context, SpoutOutputCollector collector) {// TODO Auto-generated method stubthis.map=map;this.context=context;this.collector=collector;}/*** 采集，向后发送数据* */@Overridepublic void nextTuple() {i++;List num = new Values(i);//看他的实现，这个Values就是一个可变数组，里面在不断的循环this.collector.emit(num);//把Values发出去就行了
//      this.collector.emit(num,2);//把Values发出去就行了System.err.println("spuot---------"+i);Utils.sleep(1000);}/*** 向接收数据的逻辑处理单元发送数据的字段名称* */@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("num"));//可以声明多个，取决于我向后发送了几个。类似于intent和sp的方式}}

看Values()的实现，它其实就是一个可变数组，里面在不停的循环：

public Values(Object... vals) {super(vals.length);for(Object o: vals) {add(o);}}

3.2 MyBolt 类

这个类的作用就是接受上一个Spout发送过来的数据，并求和累加。过程还是首先在prepare方法中进行初始化，然后在execute方法中根据Spout定义的发送数据字段名称，来获取到传递过来的数据。很显然这一步就能满足需求，无需再继续向后发射数据了。

public class MyBolt extends BaseRichBolt{Map stormConf;TopologyContext context;OutputCollector collector;int sum=0;@Overridepublic void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {this.stormConf=stormConf;this.context=context;this.collector=collector;}/*** 获取数据，有必要的话，向后继续发送数据* */@Overridepublic void execute(Tuple input) {//1.获取到数据int i = input.getIntegerByField("num");//因为我发过来的就是int类型，和intent、sp方式极其类似//2.求和累加sum+=i;System.err.println("sum:=============================="+sum);}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {// TODO Auto-generated method stub}}

3.3 测试类

首先要构建拓扑结构，并设置Spout、Bolt，指定分发策略。这里才用的是shuffleGrouping的分发策略。最后创建本地化集群，将我们的作业提交到集群运行即可。

public class Test {/*** 构建拓扑结构，放入集群运行* @param args命令行参数*/public static void main(String[] args) {//构建storm拓扑结构TopologyBuilder tb = new TopologyBuilder();tb.setSpout("wsspout", new MySpout());//通过shuffleGrouping将spout和bolt联系起来，指定分发策略。并行度3，写不写都行tb.setBolt("wsbolt", new MyBolt(),3).shuffleGrouping("wsspout");//创建本地storm集群LocalCluster lc = new LocalCluster();Config config = new Config();//提交作业到本地集群，拓扑作业name、作业配置信息、拓扑创建lc.submitTopology("wordsum", config, tb.createTopology());}
}