在全面介紹Storm之前，我們首先通過簡單的Demo讓我們來看看什么是整體感覺Storm。 Storm執行模式：

本地模式(Local Mode)： 即Topology（相當于一個任務，興許會具體解說） ?執行在本地機器的單一JVM上，這個模式主要用來開發、調試。

遠程模式(Remote Mode):在這個模式。我們把我們的Topology提交到集群，在這個模式中。Storm的全部組件都是線程安全的。由于它們都會執行在不同的Jvm或物理機器上，這個模式就是正式的生產模式。

寫一個HelloWord Storm ??? ?我們如今創建這么一個應用，統計文本文件里的單詞個數。具體學習過Hadoop的朋友都應該寫過。那么我們須要具體創建這樣一個Topology。用一個spout負責讀取文本文件，用第一個bolt來解析成單詞，用第二個bolt來對解析出的單詞計數。總體結構如圖所看到的： 能夠從這里下載源代碼：https://github.com/storm-book/examples-ch02-getting_started/zipball/master ??? ?寫一個可執行的Demo非常easy，我們僅僅須要三步：

創建一個Spout讀取數據

創建bolt處理數據

創建一個Topology提交到集群

以下我們就寫一下。以下代碼復制到eclipse（依賴的jar包到官網下載就可以）就可以執行。 1.創建一個Spout作為數據源 Spout作為數據源。它實現了IRichSpout接口，功能是讀取一個文本文件并把它的每一行內容發送給bolt。

package storm.demo.spout;import java.io.BufferedReader; import java.io.FileNotFoundException; import java.io.FileReader; import java.util.Map; import backtype.storm.spout.SpoutOutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.IRichSpout; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Values; public class WordReader implements IRichSpout {private static final long serialVersionUID = 1L;private SpoutOutputCollector collector;private FileReader fileReader;private boolean completed = false;public boolean isDistributed() {return false;}/*** 這是第一個方法。里面接收了三個參數。第一個是創建Topology時的配置，* 第二個是全部的Topology數據，第三個是用來把Spout的數據發射給bolt* **/@Overridepublic void open(Map conf, TopologyContext context,SpoutOutputCollector collector) {try {//獲取創建Topology時指定的要讀取的文件路徑this.fileReader = new FileReader(conf.get("wordsFile").toString());} catch (FileNotFoundException e) {throw new RuntimeException("Error reading file ["+ conf.get("wordFile") + "]");}//初始化發射器this.collector = collector;}/*** 這是Spout最基本的方法，在這里我們讀取文本文件。并把它的每一行發射出去（給bolt）* 這種方法會不斷被調用。為了減少它對CPU的消耗，當任務完畢時讓它sleep一下* **/@Overridepublic void nextTuple() {if (completed) {try {Thread.sleep(1000);} catch (InterruptedException e) {// Do nothing}return;}String str;// Open the readerBufferedReader reader = new BufferedReader(fileReader);try {// Read all lineswhile ((str = reader.readLine()) != null) {/*** 發射每一行，Values是一個ArrayList的實現*/this.collector.emit(new Values(str), str);}} catch (Exception e) {throw new RuntimeException("Error reading tuple", e);} finally {completed = true;}}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("line"));}@Overridepublic void close() {// TODO Auto-generated method stub}@Overridepublic void activate() {// TODO Auto-generated method stub}@Overridepublic void deactivate() {// TODO Auto-generated method stub}@Overridepublic void ack(Object msgId) {System.out.println("OK:" + msgId);}@Overridepublic void fail(Object msgId) {System.out.println("FAIL:" + msgId);}@Overridepublic Map<String, Object> getComponentConfiguration() {// TODO Auto-generated method stubreturn null;} } 2.創建兩個bolt來處理Spout發射出的數據 Spout已經成功讀取文件并把每一行作為一個tuple(在Storm數據以tuple的形式傳遞)發射過來。我們這里須要創建兩個bolt分別來負責解析每一行和對單詞計數。

Bolt中最重要的是execute方法，每當一個tuple傳過來時它便會被調用。

第一個bolt：WordNormalizer package storm.demo.bolt; import java.util.ArrayList; import java.util.List; import java.util.Map; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.IRichBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; public class WordNormalizer implements IRichBolt {private OutputCollector collector;@Overridepublic void prepare(Map stormConf, TopologyContext context,OutputCollector collector) {this.collector = collector;}/**這是bolt中最重要的方法，每當接收到一個tuple時。此方法便被調用* 這種方法的作用就是把文本文件里的每一行切分成一個個單詞，并把這些單詞發射出去（給下一個bolt處理）* **/@Overridepublic void execute(Tuple input) {String sentence = input.getString(0);String[] words = sentence.split(" ");for (String word : words) {word = word.trim();if (!word.isEmpty()) {word = word.toLowerCase();// Emit the wordList a = new ArrayList();a.add(input);collector.emit(a, new Values(word));}}//確認成功處理一個tuplecollector.ack(input);}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word"));}@Overridepublic void cleanup() {// TODO Auto-generated method stub}@Overridepublic Map<String, Object> getComponentConfiguration() {// TODO Auto-generated method stubreturn null;} } 第二個bolt:WordCounter package storm.demo.bolt; import java.util.HashMap; import java.util.Map; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.IRichBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Tuple;public class WordCounter implements IRichBolt {Integer id;String name;Map<String, Integer> counters;private OutputCollector collector;@Overridepublic void prepare(Map stormConf, TopologyContext context,OutputCollector collector) {this.counters = new HashMap<String, Integer>();this.collector = collector;this.name = context.getThisComponentId();this.id = context.getThisTaskId();}@Overridepublic void execute(Tuple input) {String str = input.getString(0);if (!counters.containsKey(str)) {counters.put(str, 1);} else {Integer c = counters.get(str) + 1;counters.put(str, c);}// 確認成功處理一個tuplecollector.ack(input);}/*** Topology運行完成的清理工作，比方關閉連接、釋放資源等操作都會寫在這里* 由于這僅僅是個Demo，我們用它來打印我們的計數器* */@Overridepublic void cleanup() {System.out.println("-- Word Counter [" + name + "-" + id + "] --");for (Map.Entry<String, Integer> entry : counters.entrySet()) {System.out.println(entry.getKey() + ": " + entry.getValue());}counters.clear();}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {// TODO Auto-generated method stub}@Overridepublic Map<String, Object> getComponentConfiguration() {// TODO Auto-generated method stubreturn null;} } 3.在main函數中創建一個Topology 在這里我們要創建一個Topology和一個LocalCluster對象。另一個Config對象做一些配置。

? ?

package storm.demo;import storm.demo.bolt.WordCounter; import storm.demo.bolt.WordNormalizer; import storm.demo.spout.WordReader; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.topology.TopologyBuilder; import backtype.storm.tuple.Fields; public class WordCountTopologyMain {public static void main(String[] args) throws InterruptedException {//定義一個TopologyTopologyBuilder builder = new TopologyBuilder();builder.setSpout("word-reader",new WordReader());builder.setBolt("word-normalizer", new WordNormalizer()).shuffleGrouping("word-reader");builder.setBolt("word-counter", new WordCounter(),2).fieldsGrouping("word-normalizer", new Fields("word"));//配置Config conf = new Config();conf.put("wordsFile", "d:/text.txt");conf.setDebug(false);//提交Topologyconf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);//創建一個本地模式clusterLocalCluster cluster = new LocalCluster();cluster.submitTopology("Getting-Started-Toplogie", conf,builder.createTopology());Thread.sleep(1000);cluster.shutdown();} } 執行這個函數我們就可以看到后臺打印出來的單詞個數。 （ps:由于是Local模式。執行開始可能會打印非常多錯誤log，這個先不用管）

版權聲明：本文博主原創文章，博客，未經同意不得轉載。

轉載于:https://www.cnblogs.com/blfshiye/p/4833091.html

總結

以上是生活随笔為你收集整理的Storm具体的解释（二）、成为第一Storm申请书的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。