Storm历险记之浅入浅出：Storm Hello World入门示例

import org.apache.storm.spout.SpoutOutputCollector;

import org.apache.storm.task.TopologyContext;

import org.apache.storm.topology.OutputFieldsDeclarer;

import org.apache.storm.topology.base.BaseRichSpout;

import org.apache.storm.tuple.Fields;

import org.apache.storm.tuple.Values;

public class TestSpout extends BaseRichSpout{

private static final long serialVersionUID = 225243592780939490L;

private SpoutOutputCollector collector;

private static final String field=”word”;

private int count=1;

private String[] message =  {

“Storm Hello World”,

“http://www.jishudao.com storm blog”,

“Play with storm”

};

/**

* open()方法中是在ISpout接口中定义，在Spout组件初始化时被调用。

* 有三个参数:

* 1.Storm配置的Map;

* 2.topology中组件的信息;

* 3.发射tuple的方法;

*/

public void open(Map map, TopologyContext arg1, SpoutOutputCollector collector) {

System.out.println(“open:”+map.get(“test”));

this.collector = collector;

}

/**

* nextTuple()方法是Spout实现的核心。

* 也就是主要执行方法，用于输出信息,通过collector.emit方法发射。

*/

public void nextTuple() {

if(count<=message.length){

System.out.println(“第”+count +”次开始发送数据…”);

this.collector.emit(new Values(message[count-1]));

}

count++;

}

/**

* declareOutputFields是在IComponent接口中定义，用于声明数据格式。

* 即输出的一个Tuple中，包含几个字段。

*/

public void declareOutputFields(OutputFieldsDeclarer declarer) {

System.out.println(“定义格式…”);

declarer.declare(new Fields(field));

}

/**

* 当一个Tuple处理成功时，会调用这个方法

*/

@Override

public void ack(Object obj) {

System.out.println(“ack:”+obj);

}

/**

* 当Topology停止时，会调用这个方法

*/

@Override

public void close() {

System.out.println(“关闭…”);

}

/**

* 当一个Tuple处理失败时，会调用这个方法

*/

@Override

public void fail(Object obj) {

System.out.println(“失败:”+obj);

}

}

import org.apache.storm.task.OutputCollector;

import org.apache.storm.task.TopologyContext;

import org.apache.storm.topology.OutputFieldsDeclarer;

import org.apache.storm.topology.base.BaseRichBolt;

import org.apache.storm.tuple.Fields;

import org.apache.storm.tuple.Tuple;

import org.apache.storm.tuple.Values;

public class TestBolt extends BaseRichBolt{

/**

*

*/

private static final long serialVersionUID = 4743224635827696343L;

private OutputCollector collector;

/**

* 在Bolt启动前执行，提供Bolt启动环境配置的入口

* 一般对于不可序列化的对象进行实例化。

* 注:如果是可以序列化的对象，那么最好是使用构造函数。

*/

public void prepare(Map map, TopologyContext arg1, OutputCollector collector) {

System.out.println(“prepare:”+map.get(“test”));

this.collector=collector;

}

/**

* execute()方法是Bolt实现的核心。

* 也就是执行方法，每次Bolt从流接收一个订阅的tuple，都会调用这个方法。

*/

public void execute(Tuple tuple) {

String msg=tuple.getStringByField(“word”);

System.out.println(“开始分割单词:”+msg);

String[] words = msg.toLowerCase().split(” “);

for (String word : words) {

this.collector.emit(new Values(word));//向下一个bolt发射数据

}

}

/**

* 声明数据格式

*/

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new Fields(“count”));

}

/**

* cleanup是IBolt接口中定义,用于释放bolt占用的资源。

* Storm在终止一个bolt之前会调用这个方法。

*/

@Override

public void cleanup() {

System.out.println(“TestBolt的资源释放”);

}

}

import java.util.Map;

import org.apache.storm.task.OutputCollector;

import org.apache.storm.task.TopologyContext;

import org.apache.storm.topology.OutputFieldsDeclarer;

import org.apache.storm.topology.base.BaseRichBolt;

import org.apache.storm.tuple.Tuple;

public class Test2Bolt extends BaseRichBolt{

/**

*

*/

private static final long serialVersionUID = 4743224635827696343L;

/**

* 保存单词和对应的计数

*/

private HashMap<String, Integer> counts = null;

private long count=1;

/**

* 在Bolt启动前执行，提供Bolt启动环境配置的入口

* 一般对于不可序列化的对象进行实例化。

* 注:如果是可以序列化的对象，那么最好是使用构造函数。

*/

public void prepare(Map map, TopologyContext arg1, OutputCollector collector) {

System.out.println(“prepare:”+map.get(“test”));

this.counts=new HashMap<String, Integer>();

}

/**

* execute()方法是Bolt实现的核心。

* 也就是执行方法，每次Bolt从流接收一个订阅的tuple，都会调用这个方法。

*

*/

public void execute(Tuple tuple) {

String msg=tuple.getStringByField(“count”);

System.out.println(“第”+count+”次统计单词出现的次数”);

/**

* 如果不包含该单词，说明在该map是第一次出现

* 否则进行加1

*/

if (!counts.containsKey(msg)) {

counts.put(msg, 1);

} else {

counts.put(msg, counts.get(msg)+1);

}

count++;

}

/**

* cleanup是IBolt接口中定义,用于释放bolt占用的资源。

* Storm在终止一个bolt之前会调用这个方法。

*/

@Override

public void cleanup() {

System.out.println(“===========开始显示单词数量============”);

for (Map.Entry<String, Integer> entry : counts.entrySet()) {

System.out.println(entry.getKey() + “: ” + entry.getValue());

}

System.out.println(“===========结束============”);

System.out.println(“Test2Bolt的资源释放”);

}

/**

* 声明数据格式

*/

public void declareOutputFields(OutputFieldsDeclarer arg0) {

}

}

import org.apache.storm.LocalCluster;

import org.apache.storm.StormSubmitter;

import org.apache.storm.topology.TopologyBuilder;

import org.apache.storm.tuple.Fields;

public class App {

private static final String test_spout=”test_spout”;

private static final String test_bolt=”test_bolt”;

private static final String test2_bolt=”test2_bolt”;

public static void main(String[] args)  {

//定义一个拓扑

TopologyBuilder builder=new TopologyBuilder();

//设置一个Executeor(线程)，默认一个

builder.setSpout(test_spout, new TestSpout(),1);

//shuffleGrouping:表示是随机分组

//设置一个Executeor(线程)，和一个task

builder.setBolt(test_bolt, new TestBolt(),1).setNumTasks(1).shuffleGrouping(test_spout);

//fieldsGrouping:表示是按字段分组

//设置一个Executeor(线程)，和一个task

builder.setBolt(test2_bolt, new Test2Bolt(),1).setNumTasks(1).fieldsGrouping(test_bolt, new Fields(“count”));

Config conf = new Config();

conf.put(“test”, “test”);

try{

//运行拓扑

if(args !=null&&args.length>0){ //有参数时，表示向集群提交作业，并把第一个参数当做topology名称

System.out.println(“运行远程模式”);

StormSubmitter.submitTopology(args[0], conf, builder.createTopology());

} else{//没有参数时，本地提交

//启动本地模式

System.out.println(“运行本地模式”);

LocalCluster cluster = new LocalCluster();

cluster.submitTopology(“Word-counts” ,conf,  builder.createTopology() );

Thread.sleep(20000);

//  //关闭本地集群

cluster.shutdown();

}

}catch (Exception e){

e.printStackTrace();

}

}

}