新建一个storm工程

一、maven创建工程

mvn archetype:generate -DgroupId=com.h3c.storm -DartifactId=storm-samples -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

二、导入创建好的maven工程到eclipse，修改pom文件，加上以下部分

   <dependency>

      <groupId>org.apache.storm</groupId>

      <artifactId>storm-core</artifactId>

      <version>0.10.0</version>

      <!-- keep storm out of the jar-with-dependencies -->

      <scope>provided</scope>

    </dependency>

最终的pom如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.h3c.storm</groupId>
  <artifactId>storm-samples</artifactId>
  <packaging>jar</packaging>
  <version>1.0-SNAPSHOT</version>
  <name>storm-samples</name>
  <url>http://maven.apache.org</url>
  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    
    <dependency>
      <groupId>org.apache.storm</groupId>
      <artifactId>storm-core</artifactId>
      <version>0.10.0</version>
      <!-- keep storm out of the jar-with-dependencies -->
      <scope>provided</scope>
    </dependency>
  </dependencies>
</project>

三、修改app.java 名字（当然不修改也可以）为MyFirstTopology，并写一个示例代码

package com.h3c.storm;

import java.util.Map;
import java.util.Random;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.*;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

public class MyFirstTopology {

    public static class RandomSpout extends BaseRichSpout {

        private SpoutOutputCollector collector;

        private Random rand;
        
        private static String[] sentences = new String[] {"kane1:I'm kane1", "kane2:I'm kane2", 
                                                          "kane3:I'm kane3", "kane4:I'm kane4"};
        
        @Override
        public void open(Map conf, TopologyContext context,
                SpoutOutputCollector collector) {
            this.collector = collector;
            this.rand = new Random();
        }

        @Override
        public void nextTuple() {
            String toSay = sentences[rand.nextInt(sentences.length)];
            this.collector.emit(new Values(toSay));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("sentence"));
        }

    }
    
    public static class ExclaimBasicBolt extends BaseBasicBolt {

        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
            //String sentence = tuple.getString(0);
            String sentence = (String) tuple.getValue(0);
            String out = sentence + "!";
            collector.emit(new Values(out));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("excl_sentence"));
        }

    }

    public static class PrintBolt extends BaseBasicBolt {

        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
            String rec = tuple.getString(0);
            System.err.println("String recieved: " + rec);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            // do nothing
        }

    }
    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
        
        builder.setSpout("spout", new RandomSpout());
        builder.setBolt("exclaim", new ExclaimBasicBolt()).shuffleGrouping("spout");
        builder.setBolt("print", new PrintBolt()).shuffleGrouping("exclaim");

        Config conf = new Config();
        conf.setDebug(false);

        if (args != null && args.length > 0) {
            conf.setNumWorkers(3);

            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
        } else {

            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("test", conf, builder.createTopology());
            Utils.sleep(10000);
            cluster.killTopology("test");
            cluster.shutdown();
        }
    }
}