package com.csylh;import org.apache.storm.Config;import org.apache.storm.LocalCluster;import org.apache.storm.spout.SpoutOutputCollector;import org.apache.storm.task.OutputCollector;import org.apache.storm.task.TopologyContext;import org.apache.storm.topology.OutputFieldsDeclarer;import org.apache.storm.topology.TopologyBuilder;import org.apache.storm.topology.base.BaseRichBolt;import org.apache.storm.topology.base.BaseRichSpout;import org.apache.storm.tuple.Fields;import org.apache.storm.tuple.Tuple;import org.apache.storm.tuple.Values;import org.apache.storm.utils.Utils;import java.util.Map;/** * Description:使用Storm实现累加求和操作 * * @author: 留歌36 * Date:2018/9/3 16:50 */public class LocalSumStormTopology {    /**     * Spout需要继承BaseRichSpout     * 数据源需要产生数据并发射到Bolt     */    public static class DataSourceSpout extends BaseRichSpout{        //定义一个发射器        private SpoutOutputCollector collector;        /**         * 初始化方法 只是会被调用一次         * @param conf     配置参数         * @param context 上下文：相当于一个框 可以从里面获取许多东西         * @param collector 数据发射器         */        @Override        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {            //将传入的collector发射器 对私有变量 进行赋初值            this.collector = collector;        }        int number = 0;        /**         *  用于产生数据         *  生产中肯定是从消息队列中获取数据         *  这个方法是一个死循环         */        @Override        public void nextTuple() {            //发送方式，调用上面定义的数据发射器            this.collector.emit(new Values(number++));            System.out.println("Spout==》发送的数据：" + number);            //每隔1s中发射一次，防止数据产生太快            Utils.sleep(1000);        }        /**         * 声明输出字段         * @param declarer         */        @Override        public void declareOutputFields(OutputFieldsDeclarer declarer) {            declarer.declare(new Fields("num"));        }    }    /**     * Bolt需要继承BaseRichBolt     * 用于接收数据并对数据进行处理     */    public static class SumBolt extends BaseRichBolt{        /**         *  初始化方法 ,会被执行一次         * @param stormConf         * @param context         * @param collector  这里的数据发射器,由于业务逻辑中没有没有必要进行放下发的操作,所以就是没有必要进行new一个         */        @Override        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {        }        int sum = 0;        /**         * 也是一个死循环 ,职责: 获取Spout发射过来的数据         * @param input         */        @Override        public void execute(Tuple input) {           //Bolt中获取值可以通过index获取            // 也可以根据上一个环节中定义的filed的名称获取(***推荐)          Integer value = input.getIntegerByField("num");          sum += value;          System.out.println("Bolt :Sum = ["+ sum + "]");        }        @Override        public void declareOutputFields(OutputFieldsDeclarer declarer) {        }    }    public static void main(String[] args) {        TopologyBuilder builder = new TopologyBuilder();        builder.setSpout("DataSourceSpout",new DataSourceSpout());        builder.setBolt("SumBolt",new SumBolt()).shuffleGrouping("DataSourceSpout");        //创建一个本地的Storm集群 ,本地模式运行,不需要搭建Storm集群        LocalCluster localCluster = new LocalCluster();        localCluster.submitTopology("LocalSumStormTopology",new Config(),builder.createTopology());    }}