之前已经翻译了好几篇官方的文档，项目第一阶段这两天也已经完成，按理说自己对Storm已经有一定的了解，但是由于是接手同事的工作，感觉Storm的基础没有打牢，所以自己来动手做做。
原创文章，转载请注明出处
官方推荐通过storm-starter来学习Storm，我也翻译了一篇文章：Storm(三) storm-starter，不过由于storm-starter包含多个Topology，做了一些抽象，本人觉得对初学者还是有点难度，所以决定自己新建项目。Storm(三) storm-starter这篇文章里提到了，对于初学者建议先看ExclamationTopology，我就从这个开始。如果对本文提到的Storm的概念不熟悉，建议先看Storm（二）官方Tutorial
pom文件引入依赖
既然是最简单的例子，那么依赖也是极简单的，只有一个：
<dependencies>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>1.0.1</version>
<scope>provided</scope>
</dependency>
</dependencies>
使用1.0.1版本是为了和服务器上部署的Storm集群保持一致，scope标签值设为provided，因为服务器上有运行环境，不需要生成的时候再包含。
ExclamationBolt 的实现
public static class ExclamationBolt extends BaseRichBolt {
OutputCollector _collector;
@Override
public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
_collector = collector;
}
@Override
public void execute(Tuple tuple) {
_collector.emit(tuple, new Values(tuple.getString(0) + "!!!"));
_collector.ack(tuple);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}
ExclamationBolt的代码和storm-starter里面是一致的，我没有做任何改动，这个bolt就是取出Tuple中的单词，在单词后面加"!!!"，最后emit。
main函数的实现
public static void main(String[] args) throws Exception {
Config conf = new Config();
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("word", new TestWordSpout(), 10);
builder.setBolt("exclaim1", new ExclamationBolt(), 3).shuffleGrouping("word");
builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1");
conf.setDebug(true);
String topologyName = "ExclamationTopology";
try {
LocalCluster cluster = new LocalCluster();
cluster.submitTopology(topologyName, conf,builder.createTopology());
Thread.sleep(60 * 1000);
cluster.shutdown();
} catch (Exception e) {
e.printStackTrace();
}
}
我实现的main函数和storm-starter里面ExclamationTopology类的main函数大不一样，这里没有配置文件、没有参数解析，总之更简单些。数据来源是TestWordSpout，这是storm-core库里面的类，看一下源代码就知道它只是从5个人名中每次随机的发出1个：
public void nextTuple() {
Utils.sleep(100);
final String[] words = new String[] {"nathan", "mike", "jackson", "golda", "bertels"};
final Random rand = new Random();
final String word = words[rand.nextInt(words.length)];
_collector.emit(new Values(word));
}
数据处理就是上文提到的ExclamationBolt，每次加"!!!"，然后emit。
这个Topology就是一个三个节点串在一起：word->exclaim1->exclaim2，word的并发度是10，exclaim1的并发度是3，exclaim2的并发度是2，并发度对应线程的概念。分组方式是随机分组（shuffleGrouping）。
为了简单，我们让它以local模式运行，60秒后自动停止。
打包&运行
我用的IDEA，如果是命令，可以这样打包：
mvn package
像这样提交到Storm运行：
storm jar first-topology-1.0-SNAPSHOT.jar com.quiterr.ExclamationTopology
first-topology是项目的模块名，com.quiterr是Java 包名，请做相应更改。
在IDEA中运行
Storm的local模式已经比集群模式简单多了，通常在项目开发、测试阶段使用，但是我们还想更方便一些，比如在IDEA中运行和调试Topology。只需要把依赖的scope那一行去掉即可（提供storm的本地运行环境）：
<dependencies>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>1.0.1</version>
<!--<scope>provided</scope>-->
</dependency>
</dependencies>
很简单吧，这样就可以在IDEA中使用断点之类的强大功能了。
本文源代码：https://github.com/quiterr/storm-test/tree/master/first-topology
