1 Map conf = new HashMp();
 2 //topology所有自定义的配置均放入这个Map
 3 
 4 TopologyBuilder builder = new TopologyBuilder();
 5 //创建topology的生成器
 6 
 7 int spoutParal = get("spout.parallel", 1);
 8 //获取spout的并发设置
 9 
10 SpoutDeclarer spout = builder.setSpout(SequenceTopologyDef.SEQUENCE_SPOUT_NAME,
11                 new SequenceSpout(), spoutParal);
12 //创建Spout， 其中new SequenceSpout() 为真正spout对象，SequenceTopologyDef.SEQUENCE_SPOUT_NAME 为spout的名字，注意名字中不要含有空格
13 
14 int boltParal = get("bolt.parallel", 1);
15 //获取bolt的并发设置
16 
17 BoltDeclarer totalBolt = builder.setBolt(SequenceTopologyDef.TOTAL_BOLT_NAME, new TotalCount(),
18                 boltParal).shuffleGrouping(SequenceTopologyDef.SEQUENCE_SPOUT_NAME);
19 //创建bolt， SequenceTopologyDef.TOTAL_BOLT_NAME 为bolt名字，TotalCount 为bolt对象，boltParal为bolt并发数，
20 //shuffleGrouping（SequenceTopologyDef.SEQUENCE_SPOUT_NAME）， 
21 //表示接收SequenceTopologyDef.SEQUENCE_SPOUT_NAME的数据，并且以shuffle方式，
22 //即每个spout随机轮询发送tuple到下一级bolt中
23 
24 int ackerParal = get("acker.parallel", 1);
25 Config.setNumAckers(conf, ackerParal);
26 //设置表示acker的并发数
27 
28 int workerNum = get("worker.num", 10);
29 conf.put(Config.TOPOLOGY_WORKERS, workerNum);
30 //表示整个topology将使用几个worker
31 
32 conf.put(Config.STORM_CLUSTER_MODE, "distributed");
33 //设置topolog模式为分布式，这样topology就可以放到JStorm集群上运行
34 
35 StormSubmitter.submitTopology(streamName, conf,
36                 builder.createTopology());
37 //提交topology

 1  IRichSpout{
 2 
 3     @Override
 4     public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
 5     }
 6 
 7     @Override
 8     public void close() {
 9     }
10 
11     @Override
12     public void activate() {
13     }
14 
15     @Override
16     public void deactivate() {
17     }
18 
19     @Override
20     public void nextTuple() {
21     }
22 
23     @Override
24     public void ack(Object msgId) {
25     }
26 
27     @Override
28     public void fail(Object msgId) {
29     }
30 
31     @Override
32     public void declareOutputFields(OutputFieldsDeclarer declarer) {
33     }
34 
35     @Override
36     public Map<String, Object> getComponentConfiguration() {
37         return null;
38     }

 1 IRichBolt {
 2 
 3     @Override
 4     public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
 5     }
 6 
 7     @Override
 8     public void execute(Tuple input) {
 9     }
10 
11     @Override
12     public void cleanup() {
13     }
14 
15     @Override
16     public void declareOutputFields(OutputFieldsDeclarer declarer) {
17     }
18 
19     @Override
20     public Map<String, Object> getComponentConfiguration() {
21         return null;
22     }
23 
24 }