package com.musgame.bis.chapter1.v1;

import com.musgame.bis.utils.Utils;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * 基本的storm流程
 */
public class WordCountTopology {

    private static final String SENTENCE_SPOUT_ID = "sentence-spout";
    private static final String SPLIT_BOLT_ID = "split-bolt";
    private static final String COUNT_BOLT_ID = "count-bolt";
    private static final String REPORT_BOLT_ID = "report-bolt";
    private static final String TOPOLOGY_NAME = "word-count-topology";

    public static void main(String[] args) throws Exception {

        //数据源
        SentenceSpout spout = new SentenceSpout();
        //语句分割
        SplitSentenceBolt splitBolt = new SplitSentenceBolt();
        //单词计数
        WordCountBolt countBolt = new WordCountBolt();
        //上报
        ReportBolt reportBolt = new ReportBolt();


        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout(SENTENCE_SPOUT_ID, spout);
        // SentenceSpout --> SplitSentenceBolt
        builder.setBolt(SPLIT_BOLT_ID, splitBolt)
                .shuffleGrouping(SENTENCE_SPOUT_ID);
        // SplitSentenceBolt --> WordCountBolt
        builder.setBolt(COUNT_BOLT_ID, countBolt)
                .fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
        // WordCountBolt --> ReportBolt
        builder.setBolt(REPORT_BOLT_ID, reportBolt)
                .globalGrouping(COUNT_BOLT_ID);

        Config config = new Config();

        LocalCluster cluster = new LocalCluster();

        cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());
        Utils.waitForSeconds(10);
        cluster.killTopology(TOPOLOGY_NAME);
        cluster.shutdown();
    }
}

@@ -133,24 +133,64 @@
 calCluster;%0A
+import org.apache.storm.StormSubmitter;%0A
 import org.a
@@ -1594,24 +1594,86 @@
  Config();%0A%0A
+        if(args.length %3C= 0)%0A        %7B%0A            //%E6%9C%AC%E5%9C%B0%E6%A8%A1%E5%BC%8F%0A    
         Loca
@@ -1708,24 +1708,28 @@
 Cluster();%0A%0A
+    
         clus
@@ -1805,16 +1805,20 @@
         
+   
+ 
 Utils.wa
@@ -1835,16 +1835,20 @@
 ds(10);%0A
+    
         
@@ -1884,32 +1884,36 @@
 _NAME);%0A        
+  
+  
 cluster.shutdown
@@ -1912,20 +1912,161 @@
 shutdown();%0A
+        %7Delse%0A        %7B%0A            //%E8%BF%9C%E7%A8%8B%E7%BA%BF%E4%B8%8A%E6%A8%A1%E5%BC%8F%0A            StormSubmitter.submitTopology(args%5B0%5D, config, builder.createTopology());%0A        %7D%0A
     %7D%0A%7D%0A

update example