配置好storm之后就可以开始在eclipse里面写topology了。
下面是我在网上搜到的一个简单的例子，我按照自己的理解注释了一下。
第一步，创建mvn工程
这是pom.xml文件
1 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 2 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 3 <modelVersion>4.0.0</modelVersion> 4 5 <groupId>cn.aeths.storm</groupId> 6 <artifactId>storm001</artifactId> 7 <version>0.0.1-SNAPSHOT</version> 8 <packaging>jar</packaging> 9 10 <name>storm001</name> 11 <url>http://maven.apache.org</url> 12 13 <properties> 14 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> 15 </properties> 16 17 <dependencies> 18 <dependency> 19 <groupId>org.apache.storm</groupId> 20 <artifactId>storm-core</artifactId> 21 <version>0.9.2-incubating</version> 22 <scope>provided</scope> 23 </dependency> 24 <dependency> 25 <groupId>junit</groupId> 26 <artifactId>junit</artifactId> 27 <version>3.8.1</version> 28 <scope>test</scope> 29 </dependency> 30 </dependencies> 31 32 <build> 33 <plugins> 34 <plugin> 35 <artifactId>maven-assembly-plugin</artifactId> 36 <version>2.4</version> 37 <configuration> 38 <descriptorRefs> 39 <descriptorRef>jar-with-dependencies</descriptorRef> 40 </descriptorRefs> 41 </configuration> 42 <executions> 43 <execution> 44 <id>make-assembly</id> 45 <phase>package</phase> 46 <goals> 47 <goal>single</goal> 48 </goals> 49 </execution> 50 </executions> 51 </plugin> 52 </plugins> 53 </build> 54 </project>
使用的是storm-core-0.9.2-incubating.jar
我没找到这个jar包的源码导致没看到LocalCliuster的源码，也可能是添加的方式有问题
第二步，创建spout输入tuple
这是RandomSpout.java类的代码
1 package cn.aeths.storm.storm001.spout;
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 //简单的输入spout
14 @SuppressWarnings("serial")
15 public class RandomSpout extends BaseRichSpout {
16 //继承了一个方法少的抽象类
17 //私有属性，一个收集器，收集来自spout的tuple，这个和OutputCollector的区别是可以打id保证都执行一次
18 private SpoutOutputCollector collector;
19 //这是个字符串数组，虚拟的输入源
20 private static String[] words = { "happy", "excited", "angry" };
21
22 //可以当做是worker的初始化的方法，传递的参数分别是配置信息，获取id的参数，处理流的参数。
23 public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) {
24 this.collector = arg2;
25 }
26
27 //调用这个方法就是在要求spout传递流交给收集器
28 public void nextTuple() {
29 //随机的给word赋值
30 String word = words[new Random().nextInt(words.length)];
31 //将流发射到输出，不带id的发射，所以不跟踪，没有ack和fail的调用
32 collector.emit(new Values(word));
33 }
34
35 public void declareOutputFields(OutputFieldsDeclarer arg0) {
36 //声明分发的方式
37 arg0.declare(new Fields("randomstring"));
38 }
39
40 }
第三步，创建bolt处理tuple
这是SequenceBolt.java类的代码
1 package cn.aeths.storm.storm001.bolt;
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.Tuple;
7 //这个是简单的bolt类
8 @SuppressWarnings("serial")
9 public class SenqueceBolt extends BaseBasicBolt{
10 //继承了一个BaseBasicBolt
11 //这是简单的方式，因为它也是别的类实现的BaseBasicBolt extends BaseComponent implements IBasicBolt
12 //public interface IBasicBolt extends IComponent
13 //总之继承基础的抽象类，实现了其中的两个方法
14
15 public void execute(Tuple input, BasicOutputCollector collector) {
16 //处理输入tuple或者在输入的tuple的基础上emit tuple
17 //在本例中就是获取输入，将它包装一下，打印在命令行
18 String word = (String) input.getValue(0);
19 String out = "I‘m " + word + "!";
20 System.out.println("out=" + out);
21 }
22
23 public void declareOutputFields(OutputFieldsDeclarer declarer) {
24 //声明拓扑的流的输出约束，这里没用到
25 }
26
27 }
第四步，创建topology
这是FirstTopo.java类的代码
1 package cn.aeths.storm.storm001.topology;
2
3 import backtype.storm.Config;
4 import backtype.storm.LocalCluster;
5 import backtype.storm.StormSubmitter;
6 import backtype.storm.topology.TopologyBuilder;
7 import backtype.storm.utils.Utils;
8 import cn.aeths.storm.storm001.bolt.SenqueceBolt;
9 import cn.aeths.storm.storm001.spout.RandomSpout;
10 //拓扑创建的额类
11 public class FirstTopo {
12
13 public static void main(String[] args) throws Exception {
14 //声明拓扑创建的api
15 TopologyBuilder builder = new TopologyBuilder();
16 //设置输入spout，第一个参数是id可以用这个被bolt调用，第二个参数是实例化的spout
17 builder.setSpout("spout", new RandomSpout());
18 //设置一个新的bolt处理流，第一个参数是id可以被其他的消费它的bolt使用，第二个参数是实例化的bolt
19 builder.setBolt("bolt", new SenqueceBolt()).shuffleGrouping("spout");
20 Config conf = new Config(); //方便的为拓扑创建配置，是继承map的，使用setter方法
21 conf.setDebug(false);//关闭调试模式
22 if (args != null && args.length > 0) {
23 //有参数的时候，集群模式
24 conf.setNumWorkers(3);
25 //把本类，本配置，本拓扑提交
26 StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
27 } else {
28 //本地模式
29 LocalCluster cluster = new LocalCluster();
30 //提交拓扑，名字自定义
31 cluster.submitTopology("firstTopo", conf, builder.createTopology());
32 //休眠10s
33 Utils.sleep(10000);
34 //杀死拓扑
35 cluster.killTopology("firstTopo");
36 cluster.shutdown();
37 }
38 }
39 }
第五步，提交storm运行
命令行的结果如下：
1 out=I‘m happy! 2 out=I‘m excited! 3 out=I‘m happy! 4 out=I‘m happy! 5 out=I‘m excited! 6 out=I‘m angry! 7 out=I‘m angry! 8 out=I‘m happy! 9 52008 [Thread-8-bolt] INFO backtype.storm.util - Async loop interrupted! 10 52009 [main] INFO backtype.storm.daemon.executor - Shut down executor bolt:[2 2 11 ] 12 52009 [main] INFO backtype.storm.daemon.executor - Shutting down executor spout 13 :[3 3] 14 52010 [Thread-10-spout] INFO backtype.storm.util - Async loop interrupted!
http://blog.csdn.net/annkie/article/details/6960934
这个文献教会我怎么给eclipse添加jre的源码
这是第一个本地虚拟的集群运行的topology，下面学习一下相对复杂的一个例子。
时间： 2024-10-07 16:01:01