Storm詳解二、寫第一個Storm應用

1. 本地模式(Local Mode)： 即Topology（相當於一個任務，後續會詳細講解）  運行在本地機器的單一JVM上，這個模式主要用來開發、調試。
2. 遠程模式(Remote Mode):在這個模式，我們把我們的Topology提交到集群，在這個模式中，Storm的所有組件都是線程安全的，因為它們都會運行在不同的Jvm或物理機器上，這個模式就是正式的生產模式。

1. 創建一個Spout讀取數據
2. 創建bolt處理數據
3. 創建一個Topology提交到集群

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
	 * **/
	@Override
	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一下
	 * **/
	@Override
	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;
		}

	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("line"));

	}
	@Override
	public void close() {
		// TODO Auto-generated method stub
	}
	
	@Override
	public void activate() {
		// TODO Auto-generated method stub

	}
	@Override
	public void deactivate() {
		// TODO Auto-generated method stub

	}
	@Override
	public void ack(Object msgId) {
		System.out.println("OK:" + msgId);
	}
	@Override
	public void fail(Object msgId) {
		System.out.println("FAIL:" + msgId);

	}
	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}
}

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;
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		this.collector = collector;
	}
	/**這是bolt中最重要的方法，每當接收到一個tuple時，此方法便被調用
	 * 這個方法的作用就是把文本文件中的每一行切分成一個個單詞，並把這些單詞發射出去（給下一個bolt處理）
	 * **/
	@Override
	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);
	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word"));

	}
	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}
	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}
}

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;

	@Override
	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();

	}
	@Override
	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，我們用它來打印我們的計數器
	 * */
	@Override
	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();
	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub

	}
	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}
}

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 {
		//定義一個Topology
		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(),2)
		.fieldsGrouping("word-normalizer", new Fields("word"));
		//配置
		Config conf = new Config();
		conf.put("wordsFile", "d:/text.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());
		Thread.sleep(1000);
		cluster.shutdown();
	}
}