storm入门基础实例（无可靠性保证实例）

本实例为入门篇无可靠性保证实例，关于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包：