<dependency><groupId>org.apache.storm</groupId><artifactId>storm-core</artifactId><type>jar</type><version>0.9.3-rc1</version></dependency>

 1 package com.hirain.storm.helloworld;
 2
 3 import java.util.Map;
 4 import java.util.Random;
 5
 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
13 public class SimpleSpout extends BaseRichSpout{
14
15     /**
16      *
17      */
18     private static final long serialVersionUID = 1L;
19
20     //用来发射数据的工具类
21     private SpoutOutputCollector collector;
22
23     private static String[] info = new String[]{
24         "comaple\t,12424,44w46,654,12424,44w46,654,",
25         "lisi\t,435435,6537,12424,44w46,654,",
26         "lipeng\t,45735,6757,12424,44w46,654,",
27         "hujintao\t,45735,6757,12424,44w46,654,",
28         "jiangmin\t,23545,6457,2455,7576,qr44453",
29         "beijing\t,435435,6537,12424,44w46,654,",
30         "xiaoming\t,46654,8579,w3675,85877,077998,",
31         "xiaozhang\t,9789,788,97978,656,345235,09889,",
32         "ceo\t,46654,8579,w3675,85877,077998,",
33         "cto\t,46654,8579,w3675,85877,077998,",
34         "zhansan\t,46654,8579,w3675,85877,077998,"};
35
36     Random random=new Random();
37
38
39     /**
40      * 在SpoutTracker类中被调用，每调用一次就可以向storm集群中发射一条数据（一个tuple元组），该方法会被不停的调用
41      */
42     public void nextTuple() {
43         try {
44              String msg = info[random.nextInt(11)];
45             // 调用发射方法
46             collector.emit(new Values(msg));
47             // 模拟等待100ms
48             Thread.sleep(100);
49         } catch (InterruptedException e) {
50             e.printStackTrace();
51         }
52     }
53     /**
54      * 初始化collector
55      */
56     public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
57         this.collector = collector;
58
59     }
60
61
62     /**
63      * 定义字段id，该id在简单模式下没有用处，但在按照字段分组的模式下有很大的用处。
64      * 该declarer变量有很大作用，我们还可以调用declarer.declareStream();来定义stramId，该id可以用来定义更加复杂的流拓扑结构
65      */
66     public void declareOutputFields(OutputFieldsDeclarer declarer) {
67         declarer.declare(new Fields("source")); //collector.emit(new Values(msg));参数要对应
68     }
69
70 }

 1 package com.hirain.storm.helloworld;
 2
 3 import backtype.storm.topology.BasicOutputCollector;
 4 import backtype.storm.topology.OutputFieldsDeclarer;
 5 import backtype.storm.topology.base.BaseBasicBolt;
 6 import backtype.storm.tuple.Fields;
 7 import backtype.storm.tuple.Tuple;
 8 import backtype.storm.tuple.Values;
 9
10
11
12 public class SimpleBolt extends BaseBasicBolt {
13
14     /**
15      *
16      */
17     private static final long serialVersionUID = 1L;
18
19     public void execute(Tuple input,BasicOutputCollector collector) {
20         try {
21             String msg = input.getString(0);
22             if (msg != null){
23                 //System.out.println("msg="+msg);
24                 collector.emit(new Values(msg + "msg is processed!"));
25             }
26
27         } catch (Exception e) {
28             e.printStackTrace();
29         }
30
31     }
32
33     public void declareOutputFields(
34             OutputFieldsDeclarer declarer) {
35         declarer.declare(new Fields("info"));
36
37     }
38
39 }

 1 package com.hirain.storm.helloworld;
 2
 3 import backtype.storm.Config;
 4 import backtype.storm.LocalCluster;
 5 import backtype.storm.StormSubmitter;
 6 import backtype.storm.topology.TopologyBuilder;
 7
 8 public class SimpleTopology {
 9
10
11     public static void main(String[] args) {
12         try {
13             // 实例化TopologyBuilder类。
14             TopologyBuilder topologyBuilder = new TopologyBuilder();
15             // 设置喷发节点并分配并发数，该并发数将会控制该对象在集群中的线程数。
16             topologyBuilder.setSpout("SimpleSpout", new SimpleSpout(), 1);
17             // 设置数据处理节点并分配并发数。指定该节点接收喷发节点的策略为随机方式。
18             topologyBuilder.setBolt("SimpleBolt", new SimpleBolt(), 3).shuffleGrouping("SimpleSpout");
19             Config config = new Config();
20             config.setDebug(true);
21             if (args != null && args.length > 0) {
22                 config.setNumWorkers(1);
23                 StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology());
24             } else {
25                 // 这里是本地模式下运行的启动代码。
26                 config.setMaxTaskParallelism(1);
27                 LocalCluster cluster = new LocalCluster();
28                 cluster.submitTopology("simple", config, topologyBuilder.createTopology());
29             }
30
31         } catch (Exception e) {
32             e.printStackTrace();
33         }
34     }
35 }