3. Storm编程框架

 1 package cn.itcast.storm.spout;
 2 import java.util.Map;
 3 import java.util.Random;
 4 import org.apache.commons.logging.Log;
 5 import org.apache.commons.logging.LogFactory;
 6 import backtype.storm.spout.SpoutOutputCollector;
 7 import backtype.storm.task.TopologyContext;
 8 import backtype.storm.topology.OutputFieldsDeclarer;
 9 import backtype.storm.topology.base.BaseRichSpout;
10 import backtype.storm.tuple.Fields;
11 import backtype.storm.tuple.Values;
12 import backtype.storm.utils.Utils;
13 public class RandomWordSpout extends BaseRichSpout {
14     private static final long serialVersionUID = -4287209449750623371L;
15     
16     private static final Log log = LogFactory.getLog(RandomWordSpout.class);
17     private SpoutOutputCollector collector;
18     
19     private String[] words = new String[]{"storm", "hadoop", "hive", "flume"};
20     
21     private Random random = new Random();
22     
23     public RandomWordSpout() {
24         log.warn("RandomWordSpout constructor method invoked");
25     }
26     @Override
27     public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
28         log.warn("RandomWordSpout open() method invoked");
29         this.collector = collector;
30     }
31     @Override
32     public void declareOutputFields(OutputFieldsDeclarer declarer) {
33         log.warn("RandomWordSpout declareOutputFields() method invoked");
34         declarer.declare(new Fields("str"));
35     }
36     @Override
37     public void nextTuple() {
38         log.warn("RandomWordSpout nextTuple() method invoked");
39         Utils.sleep(500);
40         String str = words[random.nextInt(words.length)];
41         collector.emit(new Values(str));
42     }
43     @Override
44     public void activate() {
45         log.warn("RandomWordSpout activate() method invoked");
46     }
47     @Override
48     public void deactivate() {
49         log.warn("RandomWordSpout deactivate() method invoked");
50     }
51 }

 1 package cn.itcast.storm.bolt;
 2 import java.util.Map;
 3 import org.apache.commons.logging.Log;
 4 import org.apache.commons.logging.LogFactory;
 5 import backtype.storm.task.TopologyContext;
 6 import backtype.storm.topology.BasicOutputCollector;
 7 import backtype.storm.topology.OutputFieldsDeclarer;
 8 import backtype.storm.topology.base.BaseBasicBolt;
 9 import backtype.storm.tuple.Fields;
10 import backtype.storm.tuple.Tuple;
11 import backtype.storm.tuple.Values;
12 public class TransferBolt extends BaseBasicBolt {
13     private static final long serialVersionUID = 4223708336037089125L;
14     private static final Log log = LogFactory.getLog(TransferBolt.class);
15     
16     public TransferBolt() {
17         log.warn("TransferBolt constructor method invoked");
18     }
19     
20     @Override
21     public void prepare(Map stormConf, TopologyContext context) {
22         log.warn("TransferBolt prepare() method invoked");
23     }
24     @Override
25     public void declareOutputFields(OutputFieldsDeclarer declarer) {
26         log.warn("TransferBolt declareOutputFields() method invoked");
27         declarer.declare(new Fields("word"));
28     }
29     @Override
30     public void execute(Tuple input, BasicOutputCollector collector) {
31         log.warn("TransferBolt execute() method invoked");
32         String word = input.getStringByField("str");
33         collector.emit(new Values(word));
34     }
35 }

package cn.itcast.storm.topology;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import cn.itcast.storm.bolt.TransferBolt;
import cn.itcast.storm.bolt.WriterBolt;
import cn.itcast.storm.spout.RandomWordSpout;
public class TopoMain {
    private static final Log log = LogFactory.getLog(TopoMain.class);
    
    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("random", new RandomWordSpout(), 2);
        builder.setBolt("transfer", new TransferBolt(), 4).shuffleGrouping("random");
        builder.setBolt("writer", new WriterBolt(), 4).fieldsGrouping("transfer", new Fields("word"));
        Config conf = new Config();
        conf.setNumWorkers(2);
        conf.setDebug(true);
        log.warn("submitting topology...");
        StormSubmitter.submitTopology("life-cycle", conf, builder.createTopology());
        log.warn("topology submitted !");
    }
}