windows下storm本地模式java開發實例

時間 2019-11-16

標籤 windows storm 本地模式 java 開發實例欄目 Windows 简体版

原文原文鏈接

環境html

storm有本地模式和集羣模式，本地模式什麼都不須要，知道storm jar包就能夠運行了，我在windows下eclipse中使用maven進行本地模式的編程。storm編程一種能夠繼承BaseRichSpout和BaseBasicBolt類，另外一種是實現IRichSpout和IRichBolt接口。java

maven下載storm jar包web

從storm官網上能夠查到storm maven配置http://storm.apache.org/downloads.htmlapache

我這裏使用的是storm-0.10.0編程

groupId: org.apache.storm
artifactId: storm-core
version: 0.10.0

第一種方法繼承類windows

編寫spoutapp

package com.storm.stormDemo;
import java.util.Map;
import java.util.Random;
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 RandomSpout extends BaseRichSpout{
    private SpoutOutputCollector collector;
    private static String[] words = {"happy","excited","angry"};
    
    /* (non-Javadoc)
     * @see backtype.storm.spout.ISpout#open(java.util.Map, backtype.storm.task.TopologyContext, backtype.storm.spout.SpoutOutputCollector)
     */
   
    public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) {
        // TODO Auto-generated method stub
        this.collector = arg2;
    }
    
    /* (non-Javadoc)
     * @see backtype.storm.spout.ISpout#nextTuple()
     */
    public void nextTuple() {
        // TODO Auto-generated method stub
        String word = words[new Random().nextInt(words.length)]; 
        collector.emit(new Values(word));
    }
    
    /* (non-Javadoc)
     * @see backtype.storm.topology.IComponent#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer)
     */
    public void declareOutputFields(OutputFieldsDeclarer arg0) {
        // TODO Auto-generated method stub
        arg0.declare(new Fields("randomstring"));
    }
}

編寫boltdom

package com.storm.stormDemo;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;
public class SenqueceBolt extends BaseBasicBolt{
    /* (non-Javadoc)
     * @see backtype.storm.topology.IBasicBolt#execute(backtype.storm.tuple.Tuple, backtype.storm.topology.BasicOutputCollector)
     */
  
    public void execute(Tuple input, BasicOutputCollector collector) {
        // TODO Auto-generated method stub
         String word = (String) input.getValue(0);  
         String out = "I'm " + word +  "!";  
         System.out.println("out=" + out);
    }
    
    /* (non-Javadoc)
     * @see backtype.storm.topology.IComponent#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer)
     */
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // TODO Auto-generated method stub
    }
}

編寫主類eclipse

package com.storm.stormDemo;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;
public class FirstTopo {
    
    public static void main(String[] args) throws Exception {  
        TopologyBuilder builder = new TopologyBuilder();   
        builder.setSpout("spout", new RandomSpout());  
        builder.setBolt("bolt", new SenqueceBolt()).shuffleGrouping("spout"); 
        Config conf = new Config();  
        conf.setDebug(false); 
        if (args != null && args.length > 0) {  
            conf.setNumWorkers(3);  
            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());  
        } else {  
  
            LocalCluster cluster = new LocalCluster();  
            cluster.submitTopology("firstTopo", conf, builder.createTopology());  
            Utils.sleep(3000);  
            cluster.killTopology("firstTopo");  
            cluster.shutdown();  
        }  
    }  
}

執行結果maven

第二種方法實現接口

編寫spout

package com.storm.stormDemo; 
  
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 
     * **/  
    public 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一下 
     * **/  
    public void nextTuple() {  
        if (completed) {  
            try {  
                Thread.sleep(1000);  
            } catch (InterruptedException e) {  
                // Do nothing  
            }  
            return;  
        }  
        String str;  
        // Open the reader  
        BufferedReader reader = new BufferedReader(fileReader);  
        try {  
            // Read all lines  
            while ((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;  
        }  
  
    }  
    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
        declarer.declare(new Fields("line"));  
  
    }  
    public void close() {  
        // TODO Auto-generated method stub  
    }  
      
    public void activate() {  
        // TODO Auto-generated method stub  
  
    }  
    public void deactivate() {  
        // TODO Auto-generated method stub  
  
    }  
    public void ack(Object msgId) {  
        System.out.println("OK:" + msgId);  
    }  
    public void fail(Object msgId) {  
        System.out.println("FAIL:" + msgId);  
  
    }  
    public Map<String, Object> getComponentConfiguration() {  
        // TODO Auto-generated method stub  
        return null;  
    }  
}

編寫第一個bolt

package com.storm.stormDemo;  
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;  
    public void prepare(Map stormConf, TopologyContext context,  
            OutputCollector collector) {  
        this.collector = collector;  
    }  
    /**這是bolt中最重要的方法，每當接收到一個tuple時，此方法便被調用 
     * 這個方法的做用就是把文本文件中的每一行切分紅一個個單詞，並把這些單詞發射出去（給下一個bolt處理） 
     * **/  
    public 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 word  
                List a = new ArrayList();  
                a.add(input);  
                collector.emit(a, new Values(word));  
            }  
        }  
        //確認成功處理一個tuple  
        collector.ack(input);  
    }  
    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
        declarer.declare(new Fields("word"));  
  
    }  
    public void cleanup() {  
        // TODO Auto-generated method stub  
  
    }  
    public Map<String, Object> getComponentConfiguration() {  
        // TODO Auto-generated method stub  
        return null;  
    }  
}

編寫第二個bolt

package com.storm.stormDemo;  
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;  
  
    public 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();  
  
    }  
    public 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);  
        }  
        // 確認成功處理一個tuple  
        collector.ack(input);  
    }  
    /** 
     * Topology執行完畢的清理工做，好比關閉鏈接、釋放資源等操做都會寫在這裏 
     * 由於這只是個Demo，咱們用它來打印咱們的計數器 
     * */  
    public 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();  
    }  
    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
        // TODO Auto-generated method stub  
  
    }  
    public Map<String, Object> getComponentConfiguration() {  
        // TODO Auto-generated method stub  
        return null;  
    }  
}

編寫主類

package com.storm.stormDemo; 
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;
public class WordCountTopologyMain {  
    public static void main(String[] args) throws InterruptedException {  
        //定義一個Topology  
        TopologyBuilder builder = new TopologyBuilder();  
        builder.setSpout("word-reader",new WordReader(),1);  
        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", "f:/test.txt");  
        conf.setDebug(false);  
        //提交Topology  
        conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);  
        //建立一個本地模式cluster  
        LocalCluster cluster = new LocalCluster();  
        cluster.submitTopology("Getting-Started-Toplogie", conf,builder.createTopology());  
        Utils.sleep(3000);  
        cluster.killTopology("Getting-Started-Toplogie");  
        cluster.shutdown();  
    }  
}

執行結果

http://www.aboutyun.com/thread-8080-1-1.html

總結

在spout讀取文件的時候，若是你設置spout的數量爲2的時候，讀取的數據就會重複，bolt就會處理重複的數據，這是咱們在編寫spout的時候要注意的。