添加gradle项目

code/src/main/java/com/raorao/storm/WordCount.java

@@ -0,0 +1,97 @@
+package com.raorao.storm;
+
+import com.raorao.storm.spout.RandomSentenceSpout;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.topology.BasicOutputCollector;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.topology.base.BaseBasicBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+
+/**
+ * 单词计数.
+ *
+ * @author Xiong Raorao
+ * @since 2018-07-29-15:35
+ */
+public class WordCount {
+
+  public static void main(String[] args) throws Exception {
+
+    TopologyBuilder builder = new TopologyBuilder();
+
+    builder.setSpout("spout", new RandomSentenceSpout(), 5);
+
+    builder.setBolt("split", new SplitSentence(), 8).shuffleGrouping("spout");
+    builder.setBolt("count", new WordCountBolt(), 12).fieldsGrouping("split", new Fields("word"));
+
+    Config conf = new Config();
+    conf.setDebug(true);
+
+    if (args != null && args.length > 0) {
+      conf.setNumWorkers(3);
+
+      StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
+    } else {
+      conf.setMaxTaskParallelism(3);
+
+      LocalCluster cluster = new LocalCluster();
+      cluster.submitTopology("word-count", conf, builder.createTopology());
+
+      Thread.sleep(10000);
+
+      cluster.shutdown();
+    }
+  }
+
+  public static class SplitSentence extends BaseBasicBolt {
+
+    public SplitSentence() {
+
+    }
+
+    @Override
+    public void execute(Tuple input, BasicOutputCollector collector) {
+      String sentence = input.getString(0);
+      String[] words = sentence.split(" ");
+      for (String word : words) {
+        collector.emit(new Values(word));
+      }
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+      declarer.declare(new Fields("word"));
+    }
+
+  }
+
+  public static class WordCountBolt extends BaseBasicBolt {
+
+    Map<String, Integer> counts = new HashMap<String, Integer>();
+
+    @Override
+    public void execute(Tuple tuple, BasicOutputCollector collector) {
+      String word = tuple.getString(0);
+      Integer count = counts.get(word);
+      if (count == null) {
+        count = 0;
+      }
+      count++;
+      counts.put(word, count);
+      collector.emit(new Values(word, count));
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+      declarer.declare(new Fields("word", "count"));
+    }
+  }
+
+}

code/src/main/java/com/raorao/storm/spout/RandomSentenceSpout.java

@@ -0,0 +1,80 @@
+package com.raorao.storm.spout;
+
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Map;
+import java.util.Random;
+
+public class RandomSentenceSpout extends BaseRichSpout {
+  private static final Logger LOG = LoggerFactory.getLogger(RandomSentenceSpout.class);
+
+  SpoutOutputCollector _collector;
+  Random _rand;
+
+
+  @Override
+  public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+    _collector = collector;
+    _rand = new Random();
+  }
+
+  @Override
+  public void nextTuple() {
+    Utils.sleep(100);
+    String[] sentences = new String[]{sentence("the cow jumped over the moon"), sentence("an apple a day keeps the doctor away"),
+        sentence("four score and seven years ago"), sentence("snow white and the seven dwarfs"), sentence("i am at two with nature")};
+    final String sentence = sentences[_rand.nextInt(sentences.length)];
+
+    LOG.debug("Emitting tuple: {}", sentence);
+
+    _collector.emit(new Values(sentence));
+  }
+
+  protected String sentence(String input) {
+    return input;
+  }
+
+  @Override
+  public void ack(Object id) {
+  }
+
+  @Override
+  public void fail(Object id) {
+  }
+
+  @Override
+  public void declareOutputFields(OutputFieldsDeclarer declarer) {
+    declarer.declare(new Fields("word"));
+  }
+
+  // Add unique identifier to each tuple, which is helpful for debugging
+  public static class TimeStamped extends RandomSentenceSpout {
+    private final String prefix;
+
+    public TimeStamped() {
+      this("");
+    }
+
+    public TimeStamped(String prefix) {
+      this.prefix = prefix;
+    }
+
+    protected String sentence(String input) {
+      return prefix + currentDate() + " " + input;
+    }
+
+    private String currentDate() {
+      return new SimpleDateFormat("yyyy.MM.dd_HH:mm:ss.SSSSSSSSS").format(new Date());
+    }
+  }
+}