本实例为入门篇无可靠性保证实例，关于storm的介绍，以及一些术语名词等，可以参考Storm介绍(一)、Storm介绍(二)。
本案例是基于storm0.9.3版本
1.案例结构
案例：Word Count案例
语句Spout --> 语句分隔Bolt --> 单词计数Bolt --> 上报Bolt
2.语句生成Spout - SentenceSpout
作为入门案例，我们直接从一个数组中不断读取语句，作为数据来源。
SentenceSpout不断读取语句将其作为数据来源，组装成单值tuple（键名sentence，键值为祖父穿格式的语句）向后发射。
{"sentence":"i am so shuai!"}
3.代码结构
话不多说，上代码：
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields; public class WCTopologyDriver {
public static void main(String[] args) throws Exception {
//1.创建组件
SentenceSpout sentenceSpout = new SentenceSpout();
SplitSentenceBolt splitSentenceBolt = new SplitSentenceBolt();
WordCountBolt wordCountBolt = new WordCountBolt();
ReportBolt reportBolt = new ReportBolt(); //2.创建构建者
TopologyBuilder builder = new TopologyBuilder(); //3.向构建者描述拓扑结构
builder.setSpout("Sentence_Spout", sentenceSpout);
builder.setBolt("Split_Sentence_Bolt", splitSentenceBolt)
.shuffleGrouping("Sentence_Spout");
builder.setBolt(" ", wordCountBolt)
.fieldsGrouping("Split_Sentence_Bolt", new Fields("word"));
builder.setBolt("Report_Bolt", reportBolt)
.shuffleGrouping("Word_Count_Bolt"); //4.通过构建者创建拓扑
StormTopology topology = builder.createTopology(); //5.将拓扑提交到集群中运行
//Config conf = new Config();
//StormSubmitter.submitTopology("WC_Topology", conf, topology); //5.创建本地集群 模拟运行拓扑
LocalCluster cluster = new LocalCluster();
Config conf = new Config();
cluster.submitTopology("WC_Topology", conf, topology); Thread.sleep(10 * 1000);
cluster.killTopology("WC_Topology");
cluster.shutdown();
}
}
import java.util.Map; import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values; public class SentenceSpout extends BaseRichSpout { private String [] sentences = {
"my name is park",
"i am so shuai",
"do you like me",
"are you sure you do not like me",
"ok i am sure"
}; private SpoutOutputCollector collector = null; /**
* 初始化的方法
* 当前组件初始化时 调用 执行初始化操作
* conf:代表当前topology相关配置信息
* context:代表上下文环境 可以用来获取 任务id 组件id 输入输出相关信息 等信息
* collector:代表发送者 可以用来发送 拓扑 可以在任何时候发送 此对象线程安全 可以放心的保存在类的内部作为类的成员
*/
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
} /**
* storm会在一个单一线程中不停的调用此方法 要求发送tuple
* 如果有数据要发 直接发 如果没有数据要发 也不要阻塞这个方法 而是直接返回即可
* 如果真的没有数据要发送 最好睡上一个很短的时间 以便释放cpu 不至于浪费过多资源
*/
private int index = 0;
@Override
public void nextTuple() {
if(index < sentences.length){
collector.emit(new Values(sentences[index]));
index++;
}else{
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return;
}
} /**
* 用来声明输出信息
* declarer:声明输出的流的编号 输出的tuple中的字段 以及是否是一个指向性的流
* 要注意 组件发送的tuple的结构 都要现在此方法中声明
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("sentence"));
} }
import java.util.Map; import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values; public class SplitSentenceBolt extends BaseRichBolt{ private OutputCollector collector = null; /**
* 初始化的方法
* 当前组件初始化时 调用 执行初始化操作
* conf:代表当前topology相关配置信息
* context:代表上下文环境 可以用来获取 任务id 组件id 输入输出相关信息 等信息
* collector:代表发送者 可以用来发送 拓扑 可以在任何时候发送 此对象线程安全 可以放心的保存在类的内部作为类的成员
*/
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
} /**
* 对于输入的tuple 一个tuple触发一次此方法
* 在这个方法中对tuple进行处理
*/
@Override
public void execute(Tuple input) {
String sentence = input.getStringByField("sentence");
String [] words = sentence.split(" ");
for(String word : words){
collector.emit(new Values(word));
}
} /**
* 用来声明输出信息
* declarer:声明输出的流的编号 输出的tuple中的字段 以及是否是一个指向性的流
* 要注意 组件发送的tuple的结构 都要现在此方法中声明
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
} }
import java.util.HashMap;
import java.util.Map; import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values; public class WordCountBolt extends BaseRichBolt { private OutputCollector collector = null; @Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
} private Map<String,Integer> map = new HashMap<>();
@Override
public void execute(Tuple input) {
String word = input.getStringByField("word");
map.put(word, map.containsKey(word) ? map.get(word)+1 : 1);
collector.emit(new Values(word,map.get(word)));
} @Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word","count"));
} }
import java.util.Map; import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple; public class ReportBolt extends BaseRichBolt { @Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { } @Override
public void execute(Tuple input) {
String word = input.getStringByField("word");
int count = input.getIntegerByField("count");
System.out.println("--单词数量发生变化："+word+"~"+count+"--");
} @Override
public void declareOutputFields(OutputFieldsDeclarer declarer) { } }
运行结果：
补充，以下是本文案例用到的jar包，由于太大，没有上传，下载0.9.3的storm源码，解压后文件夹中的lib下的所有jar包：