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;
import backtype.storm.utils.Utils;
import org.joda.time.DateTime;import java.util.Map;/*** 创建数据*/
public class CreateSpout extends BaseRichSpout {private SpoutOutputCollector collector;private String[] sentences = null; //用来存放数据@Overridepublic void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {this.collector = spoutOutputCollector;sentences = new String[]{"Hahahaha! I am coming ~~"}; // "Hahaha...这个字符串即为要源源不断发送的信息"}@Overridepublic void nextTuple() {/*storm会循环的调用这个方法*//*线程进行休眠,10s发送一次数据,在这10s内,让其余工作进行*/Utils.sleep(10000);//获得数据源System.out.println(new DateTime().toString("HH:mm:ss") + "--------------CreateSpout 开始发送数据----------");this.collector.emit(new Values(sentences)); //发送出去}@Overridepublic void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
//发送时设置接收方的Tuple实例中的keyoutputFieldsDeclarer.declare(new Fields("sentence"));}}

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;
import org.joda.time.DateTime;import java.util.Map;/*** 按照空格切分字符串,然后推出去*/
public class SplitBolt extends BaseRichBolt {private OutputCollector outputCollector;private int countTime = 0;@Overridepublic void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {this.outputCollector = collector;}@Overridepublic void execute(Tuple tuple) {/*这是官方文档中 tuple.getString(int i)的解释:Returns the String at position i in the tuple. If that field is not a String, you will get a runtime error.public String getString ( int i);*/String sentence = tuple.getString(0);for (String word : sentence.split(" ")) {System.out.println(new DateTime().toString("HH:mm:ss") + "--------------------SplitBolt 开始运行--------------------\n" + "> > > >  第"+count() +"次发送数据,这次发送的是:" + word);outputCollector.emit(new Values(word));}}//这是为了得到当前一共发送了多少个单词了,加深理解private int count() {return ++countTime;}/*在发射的时候,将接收方的tuple中的 key 设置为"word"*/@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word"));}
}

import backtype.storm.Config;
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;
import backtype.storm.utils.TupleHelpers;
import org.joda.time.DateTime;import java.util.HashMap;
import java.util.Map;public class CountBolt extends BaseRichBolt {private Map<String, Integer> counts = new HashMap<>();private OutputCollector outputCollector;@Overridepublic void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {this.outputCollector = collector;}@Overridepublic Map<String, Object> getComponentConfiguration() {Map<String, Object> conf = new HashMap<String, Object>();/*加入Tick时间窗口, 统计*//*------------------?????????????????????????---------------------------*/conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 10);return conf;}@Overridepublic void execute(Tuple tuple) {/*时间窗口定义为10s内的统计数据,统计完毕后,发射到下一阶段的bolt进行处理*///发射完成后return结束,开始新一轮的事件窗口计数操作if (TupleHelpers.isTickTuple(tuple)) {/*来判断是否应该发射当前窗口数据*/System.out.println((new DateTime().toString("HH:mm:ss")) + "--------------------sumWordBolt 开始运行--------------------\n发送的数据内容是" + counts);outputCollector.emit(new Values(counts));return;}/*如果没有到发送时间,就继续统计wordcount*/String word = tuple.getStringByField("word");Integer count = counts.get(word);if (count == null) {count = 0;}count++;counts.put(word, count);}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word_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;
import org.joda.time.DateTime;import java.util.Map;public class PrintBolt extends BaseRichBolt {@Overridepublic void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {}@Overridepublic void execute(Tuple input) {System.out.println(new DateTime().toString("HH:mm:ss") + "--------------------final bolt 开始运行--------------------");/*----------???????????????????????---------------------------*/Map<String, Integer> counts = (Map<String, Integer>) input.getValue(0);/*最后一个阶段,将最后的结果打印出来*/System.out.println(justForm(20-8)+"key"+justForm(20-8)+"      "+"value");for (Map.Entry<String, Integer> kv : counts.entrySet()) {/*这里的justForm()函数是为了保证格式一致*/System.out.println(kv.getKey() + justForm(kv.getKey().length()) + " 频数 : " + kv.getValue());}}//保证格式一致的私有方法private String justForm(int length) {for (int i = 0; i < 20 - length; i++) {System.out.print(" ");}return "";}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {}
}