转载请注明出处：http://blog.csdn.net/lonelytrooper/article/details/9960723

第二章 开始

在本章中，我们会建一个storm工程和我们的第一个storm topology。

 下述假设你安装了至少1.6版本的Java运行时环境(JRE)。我们推荐使用oracle提供的JRE，可以在这里找到http://www.java.com/downloads/。

操作模式

在我们开始之前，理解storm的操作模式很重要。有两种方式运行storm。

本地模式

在本地模式中，storm topologies运行在本地机器一个单独的JVM中。由于是最简单的查看所有的topology组件一起工作的模式，这种方式被用来开发，测试和调试。在这种模式下，我们可以调整参数，这使得我们可以看到我们的topology在不同的storm配置环境下是怎么运行的。为了以本地模式运行topologies，我们需要下载storm的开发依赖包，其中包含了我们开发和测试topology所需的所有东西。

当我们建立自己的第一个storm工程的时候我们很快就可以看到是怎么回事了。

在本地模式运行topology与在storm集群中运行它是类似的。然而，确保所有的组件线程安全是重要的，因为当它们被部署到远程模式中时，它们可能运行在不同的JVM中或者在不同的物理机器上，这样的话，它们之间没有直接的交流或者内存共享。

本章的所有示例，我们都以本地模式运行。

远程模式

在远程模式中，我们提交topology到storm集群，该集群由许多进程组成，通常运行在不同的机器上。远程模式不显示调试信息，这也是它被认为是生产模式的原因。然而，在一台单独的开发机器上建立storm集群是可能的，并且它被认为是在部署至生产前的一个好方法，这可以确保在生产环境中运行topology时没有任何问题。

你在第六章中可以了解到更多关于远程模式的内容，我会在附录B里展示怎样安装一个集群。

Hello World Storm

在这个工程中，我们会建立一个简单的topology来为单词计数。我们可以把这个工程

认为是storm topologies的“hello world”。然而，它是一个非常强大的topology，因为

它只需要做一些小的改动便可以扩展到几乎无限规模并且，我们甚至可以用它来做一个统计系统。例如，我们可以修改这个项目来找出Twitter上的话题趋势。

为了建立这个topology，我们将使用一个spout来负责读取单词，第一个bolt来标准化单词，

第二个bolt来为单词计数，正如我们在图2-1中看到的那样。

图2-1 开始topology

你可以在https://github.com/storm-book/examples-ch02-getting_started/zipball/master

下载示例源代码的ZIP文件。

如果你使用git(一个分布式的校正控制及源代码管理工具)，你可以在你想

要下载的源代码的目录中运行git clone git@github.com:storm-book/examplesch02-

getting_started.git。

验证Java安装

安装环境的第一步是验证你正在运行的java的版本。打开一个终端窗口，运行命令java –version。我们可以看到如下类似的信息：

 

java -version

java version "1.6.0_26"

Java(TM) SE Runtime Environment(build 1.6.0_26-b03)

Java HotSpot(TM) Server VM (build 20.1-b02,mixed mode)

如果没有，验证下你的java安装。(见http://www.java.com/download/.)

创建工程

为开始这个工程，先建立一个用来存放应用的文件夹(就像对任何的java应用一样)。该文件夹包含工程的源代码。

接着我们需要下载storm的依赖包：一个我们将添加到应用类路径的jar包的集合。你可以用两种方式中的一种做这件事：

﹒下载依赖包，解压，添加到类路径。

﹒使用ApacheMaven

maven是一套软件工程管理的工具。它可以被用来管理软件开发周期中的多个方面，从依赖到发布构建过程。在本书中我们会广泛的使用它。为验证是否已安装了maven，运行命令mvn。如果没有，可以从http://maven.apache.org/download.html下载。尽管使用storm没有必要成为一个maven专家，但是知道maven是怎样工作的基础知识是有帮助的。你可以找到更多信息在Apache Maven的网站(http://maven.apache.org/)。

为了定义工程的结构，我们需要建立一个pom.xml(工程对象模型)文件，该文件描述依赖，包，源码等。我们将使用依赖包及nathanmarz建立的maven库(https://github.com/nathan

marz/)。这些依赖可以在这里找到https://github.com/nathanmarz/storm/wiki/Maven。

storm的maven依赖包引用了在本地模式运行storm所需的所有库函数。

使用这些依赖包，我们可以写一个包含运行topology基本的必要组件的pom.xml文件：

<projectxmlns="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/xsd/maven-4.0.0.xsd">

<modelVersion>4.0.0</modelVersion>

<groupId>storm.book</groupId>

<artifactId>Getting-Started</artifactId>

<version>0.0.1-SNAPSHOT</version>

 

<build>

<plugins>

<plugin>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-compiler-plugin</artifactId>

<version>2.3.2</version>

<configuration>

<source>1.6</source>

<target>1.6</target>

<compilerVersion>1.6</compilerVersion>

</configuration>

</plugin>

</plugins>

</build>

 

<repositories>

 

<!-- Repository where we can found the storm dependencies -->

<repository>

<id>clojars.org</id>

<url>http://clojars.org/repo</url>

</repository>

</repositories>

 

<dependencies>

 

<!-- Storm Dependency -->

<dependency>

<groupId>storm</groupId>

<artifactId>storm</artifactId>

<version>0.6.0</version>

</dependency>

</dependencies>

</project>

前几行指定了工程的名字和版本。然后我们添加了一个编译器插件，该插件告诉maven我们的代码应该用Java1.6编译。接下来我们定义库(maven支持同一工程的多个库)。Clojars是storm依赖包所在的库。Maven会自动下载本地模式运行storm所需的所有子依赖包。

应用有如下的结构，典型的maven java工程：

Java下的文件夹包含我们的源代码并且我们会将我们的单词文件放到resources文件夹中来处理。

mkdir –p 建立所有所需的父目录。

建立我们第一个Topology

为建立我们第一个topology，我们要创建运行单词计数的所有的类。或许示例的一些部分在目前阶段不是很清晰，我们将在后边的章节中解释它们。

Spout

WordReader spout是实现了IRichSpout接口的类。我们在第四章会看到更多的细节。WordReader负责读文件并且将每行提供给一个bolt。

一个spout发射一个定义的域的列表。这个架构允许你有多种bolt读取相同的spout流，然后这些bolt定义域供其他的bolt消费等等。

示例2-1包含了这个类的完整代码(我们在示例后分析代码的每个部分)。

Example 2-1.src/main/java/spouts/WordReader.java

package spouts;

import java.io.BufferedReader;

import java.io.FileNotFoundException;

importjava.io.FileReader;

import java.util.Map;

import backtype.storm.spout.SpoutOutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.IRichSpout;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

 

public class WordReaderimplementsIRichSpout{

private SpoutOutputCollector collector;

private FileReader fileReader;

private booleancompleted=false;

private TopologyContext context;

public booleanisDistributed(){returnfalse;}

public voidack(Object msgId) {

System.out.println("OK:"+msgId);

}

public voidclose(){}

public voidfail(Object msgId) {

System.out.println("FAIL:"+msgId);

}

/**

* The only thing that the methods will do It is emit each

* file line

*/

public voidnextTuple() {

/**

* The nextuple it is called forever, so if we have beenreaded the file

* we will wait and then return

*/

if(completed){

try {

Thread.sleep(1000);

} catch(InterruptedExceptione) {

//Do nothing

}

return;

}

String str;

//Open the reader

BufferedReader reader =newBufferedReader(fileReader);

try{

//Read all lines

while((str=reader.readLine())!=null){

/**

* By each line emmit a new value with the line as a their

*/

this.collector.emit(newValues(str),str);

}

}catch(Exception e){

throw new RuntimeException("Errorreading tuple",e);

}finally{

completed = true;

                   }

}

/**

* We will create the file and get the collector object

*/

public voidopen(Map conf,TopologyContextcontext,

SpoutOutputCollector collector) {

try {

this.context=context;

this.fileReader=newFileReader(conf.get("wordsFile").toString());

} catch(FileNotFoundExceptione) {

throw new RuntimeException("Errorreading file

["+conf.get("wordFile")+"]");

}

this.collector=collector;

}

/**

* Declare the output field "word"

*/

public voiddeclareOutputFields(OutputFieldsDeclarerdeclarer) {

declarer.declare(newFields("line"));

}       

}

在任何spout中都调用的第一个方法是void open(Map conf, TopologyContext

context, SpoutOutputCollector collector)。方法的参数是TopologyContext，它包含了所有的topology数据；conf对象，它在topology定义的时候被创建；SpoutOutputCollector，它使得我们可以发射将被bolt处理的数据。下面的代码是open方法的实现：

public voidopen(Map conf,TopologyContext context,

SpoutOutputCollector collector) {

try {

this.context=context;

this.fileReader=newFileReader(conf.get("wordsFile").toString());

} catch(FileNotFoundException e) {

throw new RuntimeException("Error reading file ["+conf.get("wordFile")+"]");

}

this.collector=collector;

}

在这个方法中，我们也创建了reader，它负责读文件。接着我们需要实现public void nextTuple()，在这个方法里我们发射将被bolt处理的值。在我们的例子中，这个方法读文件并且每行发射一个值。

public voidnextTuple(){

if(completed){

try {

Thread.sleep(1);

} catch(InterruptedException e) {

//Do nothing

}

return;

}

String str;

BufferedReader reader =newBufferedReader(fileReader);

try{

while((str=reader.readLine())!=null){

this.collector.emit(newValues(str));

}

}catch(Exception e){

throw new RuntimeException("Errorreading tuple",e);

}finally{

completed =true;

}

}

Values是ArrayList的一个实现，其中把list的元素传到了构造方法中。

 

nextTuple()方法在相同的循环中被周期性的调用，正如ack()和fail()方法。当没有工作要做时，必须释放对线程的控制这样其他的方法有机会被调用。所以nextTuple方法的第一行是检查处理是否完成了。如果已经完成，在返回前它会休眠至少一毫秒来降低处理器的负载。如果有工作要做，那么文件的每一行被读取为一个值并且发射。

元组(Tuple)是一个值的命名列表，它可以是任何类型的java对象(只要这个对象是可序列化的)。Storm在缺省的情况下可以序列化常用的类型例如strings，bytearrays，ArrayList，HashMap和HashSet。

转自：http://blog.csdn.net/u011689674/article/details/9960723