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; //用来存放数据


    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.collector = spoutOutputCollector;
        sentences = new String[]{"Hahahaha! I am coming ~~"}; // "Hahaha...这个字符串即为要源源不断发送的信息"
    }

    @Override
    public void nextTuple() {
        /*storm会循环的调用这个方法*/
        /*线程进行休眠,10s发送一次数据,在这10s内,让其余工作进行*/
        Utils.sleep(10000);
        //获得数据源
        System.out.println(new DateTime().toString("HH:mm:ss") + "--------------CreateSpout 开始发送数据----------");
        this.collector.emit(new Values(sentences)); //发送出去
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
//发送时设置接收方的Tuple实例中的key
        outputFieldsDeclarer.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;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.outputCollector = collector;
    }

    @Override
    public 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"*/
    @Override
    public 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;


    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.outputCollector = collector;
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        Map<String, Object> conf = new HashMap<String, Object>();
        /*加入Tick时间窗口, 统计*/
        /*------------------?????????????????????????---------------------------*/
        conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 10);
        return conf;
    }

    @Override
    public 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);
    }

    @Override
    public 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 {
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
    }

    @Override
    public 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 "";
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }
}