【转】storm 开发系列一 第一个程序

2021-12-20 23:16:47

原文： http://blog.csdn.net/csfreebird/article/details/49104777

-------------------------------------------------------------------------------------------------

本文将在本地开发环境创建一个storm程序，力求简单。

首先用mvn创建一个简单的工程hello_storm

  1. mvn archetype:generate -DgroupId=org.csfreebird -DartifactId=hello_storm -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

编辑pom.xml，添加dependency

  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/maven-v4_0_0.xsd">
  3. <modelVersion>4.0.0</modelVersion>
  4. <groupId>org.csfreebird</groupId>
  5. <artifactId>hello_storm</artifactId>
  6. <version>0.9.5</version>
  7. <packaging>jar</packaging>
  8. <name>hello_storm</name>
  9. <url>http://maven.apache.org</url>
  10. <dependencies>
  11. <dependency>
  12. <groupId>org.apache.storm</groupId>
  13. <artifactId>storm-core</artifactId>
  14. <version>${project.version}</version>
  15. <!-- keep storm out of the jar-with-dependencies -->
  16. <scope>provided</scope>
  17. </dependency>
  18. </dependencies>
  19. </project>

provided 表示storm-core的jar包只作为编译和测试时使用，在集群环境下运行时完全依赖集群环境的storm-core的jar包。

然后重命名App.Java为HelloTopology.java文件，开始编码。模仿之前的Example, 这里将所有的spout/bolt类都作为静态类定义，就放在HelloTopology.java文件。

功能如下

【转】storm 开发系列一 第一个程序

编写HelloTopology.java代码，spout代码来自于TestWordSpout,去掉了log的代码，改变了_引导的成员变量命名方法

  1. package org.csfreebird;
  2. import backtype.storm.Config;
  3. import backtype.storm.LocalCluster;
  4. import backtype.storm.StormSubmitter;
  5. import backtype.storm.task.OutputCollector;
  6. import backtype.storm.task.TopologyContext;
  7. import backtype.storm.testing.TestWordSpout;
  8. import backtype.storm.topology.OutputFieldsDeclarer;
  9. import backtype.storm.topology.TopologyBuilder;
  10. import backtype.storm.topology.base.BaseRichBolt;
  11. import backtype.storm.topology.base.BaseRichSpout;
  12. import backtype.storm.tuple.Fields;
  13. import backtype.storm.tuple.Tuple;
  14. import backtype.storm.tuple.Values;
  15. import backtype.storm.utils.Utils;
  16. import backtype.storm.spout.SpoutOutputCollector;
  17. import java.util.Map;
  18. import java.util.TreeMap;
  19. import java.util.Random;
  20. public class HelloTopology {
  21. public static class HelloSpout extends BaseRichSpout {
  22. boolean isDistributed;
  23. SpoutOutputCollector collector;
  24. public HelloSpout() {
  25. this(true);
  26. }
  27. public HelloSpout(boolean isDistributed) {
  28. this.isDistributed = isDistributed;
  29. }
  30. public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
  31. this.collector = collector;
  32. }
  33. public void close() {
  34. }
  35. public void nextTuple() {
  36. Utils.sleep(100);
  37. final String[] words = new String[] {"china", "usa", "japan", "russia", "england"};
  38. final Random rand = new Random();
  39. final String word = words[rand.nextInt(words.length)];
  40. this.collector.emit(new Values(word));
  41. }
  42. public void ack(Object msgId) {
  43. }
  44. public void fail(Object msgId) {
  45. }
  46. public void declareOutputFields(OutputFieldsDeclarer declarer) {
  47. declarer.declare(new Fields("word"));
  48. }
  49. @Override
  50. public Map<String, Object> getComponentConfiguration() {
  51. if(!this.isDistributed) {
  52. Map<String, Object> ret = new TreeMap<String, Object>();
  53. ret.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 1);
  54. return ret;
  55. } else {
  56. return null;
  57. }
  58. }
  59. }
  60. public static class HelloBolt extends BaseRichBolt {
  61. OutputCollector collector;
  62. @Override
  63. public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
  64. this.collector = collector;
  65. }
  66. @Override
  67. public void execute(Tuple tuple) {
  68. this.collector.emit(tuple, new Values("hello," + tuple.getString(0)));
  69. this.collector.ack(tuple);
  70. }
  71. @Override
  72. public void declareOutputFields(OutputFieldsDeclarer declarer) {
  73. declarer.declare(new Fields("word"));
  74. }
  75. }
  76. public static void main(String[] args) throws Exception {
  77. TopologyBuilder builder = new TopologyBuilder();
  78. builder.setSpout("a", new HelloSpout(), 10);
  79. builder.setBolt("b", new HelloBolt(), 5).shuffleGrouping("a");
  80. Config conf = new Config();
  81. conf.setDebug(true);
  82. if (args != null && args.length > 0) {
  83. conf.setNumWorkers(3);
  84. StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
  85. } else {
  86. String test_id = "hello_test";
  87. LocalCluster cluster = new LocalCluster();
  88. cluster.submitTopology(test_id, conf, builder.createTopology());
  89. Utils.sleep(10000);
  90. cluster.killTopology(test_id);
  91. cluster.shutdown();
  92. }
  93. }
  94. }

编译成功

  1. mvn clean compile

为了能够在本地模式运行，需要在pom.xml中添加如下：

  1. <build>
  2. <plugins>
  3. <plugin>
  4. <groupId>org.codehaus.mojo</groupId>
  5. <artifactId>exec-maven-plugin</artifactId>
  6. <version>1.2.1</version>
  7. <executions>
  8. <execution>
  9. <goals>
  10. <goal>exec</goal>
  11. </goals>
  12. </execution>
  13. </executions>
  14. <configuration>
  15. <executable>java</executable>
  16. <includeProjectDependencies>true</includeProjectDependencies>
  17. <includePluginDependencies>false</includePluginDependencies>
  18. <classpathScope>compile</classpathScope>
  19. <mainClass>${storm.topology}</mainClass>
  20. </configuration>
  21. </plugin>
  22. </plugins>
  23. </build>

然后运行命令

    1. mvn compile exec:java -Dstorm.topology=org.csfreebird.HelloTopology
上一篇：20160510--hibernate懒加载问题


下一篇：parseInt(),parseFloat(),parse()