流式处理框架storm浅析（下篇）

整个topology分为三个部分：

WordReader：数据源，负责发送sentence

WordNormalizer：负责将sentence切分

Wordcounter:负责对单词的频率进行累加

3 代码实现

    1. 构建maven环境，添加storm依赖

	<repositories>  
	      <!-- Repository where we can found the storm dependencies  -->  
	      <repository>  
	          <id>clojars.org</id>  
	          <url>http://clojars.org/repo</url>  
	      </repository>  
	</repositories>  
	<dependencies>  
	      <dependency>   
	        <groupId>storm</groupId>  
	        <artifactId>storm</artifactId>  
	        <version>0.7.1</version>  
	     </dependency>  
	</dependencies>  

    2. 定义Topology

	public class TopologyMain {  
	    public static void main(String[] args) throws InterruptedException {  
	           
	        //Topology definition  
	        TopologyBuilder builder = new TopologyBuilder();  
	        builder.setSpout("word-reader",new WordReader());  
	        builder.setBolt("word-normalizer", new WordNormalizer())  
	            .shuffleGrouping("word-reader");  
	        builder.setBolt("word-counter", new WordCounter(),1)  
	            .fieldsGrouping("word-normalizer", new Fields("word"));  
	          
	        //Configuration  
	        Config conf = new Config();  
	        conf.put("wordsFile", args[0]);  
	        conf.setDebug(false);  
	        //Topology run  
	        conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);  
	        LocalCluster cluster = new LocalCluster();  
	        cluster.submitTopology("Getting-Started-Toplogie", conf, builder.createTopology());  
	        Thread.sleep(1000);  
	        cluster.shutdown();  
	    }  
	}  

    3. 实现WordReader Spout

	public class WordReader extends BaseRichSpout {  
	  
	    private SpoutOutputCollector collector;  
	    private FileReader fileReader;  
	    private boolean completed = false;  
	    public void ack(Object msgId) {  
	        System.out.println("OK:"+msgId);  
	    }  
	    public void close() {}  
	    public void fail(Object msgId) {  
	        System.out.println("FAIL:"+msgId);  
	    }  
	  
	    public void nextTuple() {  
	        if(completed){  
	            try {  
	                Thread.sleep(1000);  
	            } catch (InterruptedException e) {  
	            }  
	            return;  
	        }  
	        String str;  
	        BufferedReader reader = new BufferedReader(fileReader);  
	        try{  
	            while((str = reader.readLine()) != null){  
	                this.collector.emit(new Values(str),str);  
	            }  
	        }catch(Exception e){  
	            throw new RuntimeException("Error reading tuple",e);  
	        }finally{  
	            completed = true;  
	        }  
	    }  
	  
	    public void open(Map conf, TopologyContext context,  
	                     SpoutOutputCollector collector) {  
	        try {  
	            this.fileReader = new FileReader(conf.get("wordsFile").toString());  
	        } catch (FileNotFoundException e) {  
	            throw new RuntimeException("Error reading file ["+conf.get("wordFile")+"]");  
	        }  
	        this.collector = collector;  
	    }  
	  
	    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
	        declarer.declare(new Fields("line"));  
	    }  
	} 

第一个被调用的spout方法都是public void open(Map conf, TopologyContext context, SpoutOutputCollector collector)。它接收如下参数：配置对象，在定义topology对象是创建；TopologyContext对象，包含所有拓扑数据；还有SpoutOutputCollector对象，它能让我们发布交给bolts处理的数据。

	public class WordNormalizer extends BaseBasicBolt {  
	  
	    public void cleanup() {}  
	  
	    public void execute(Tuple input, BasicOutputCollector collector) {  
	        String sentence = input.getString(0);  
	        String[] words = sentence.split(" ");  
	        for(String word : words){  
	            word = word.trim();  
	            if(!word.isEmpty()){  
	                word = word.toLowerCase();  
	                collector.emit(new Values(word));  
	            }  
	        }  
	    }  
	      
	    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
	        declarer.declare(new Fields("word"));  
	    }  
	}  

	public class WordCounter extends BaseBasicBolt {  
	  
	    Integer id;  
	    String name;  
	    Map counters;  
	  
	    @Override  
	    public void cleanup() {  
	        System.out.println("-- Word Counter ["+name+"-"+id+"] --");  
	        for(Map.Entry entry : counters.entrySet()){  
	            System.out.println(entry.getKey()+": "+entry.getValue());  
	        }  
	    }  
	  
	    @Override  
	    public void prepare(Map stormConf, TopologyContext context) {  
	        this.counters = new HashMap();  
	        this.name = context.getThisComponentId();  
	        this.id = context.getThisTaskId();  
	    }  
	  
	    @Override  
	    public void declareOutputFields(OutputFieldsDeclarer declarer) {}  
	  
	    @Override  
	    public void execute(Tuple input, BasicOutputCollector collector) {  
	        String str = input.getString(0);  
	        if(!counters.containsKey(str)){  
	            counters.put(str, 1);  
	        }else{  
	            Integer c = counters.get(str) + 1;  
	            counters.put(str, c);  
	        }  
	    }  
	} 

• 附-Storm记录级容错的基本原理