Apache Storm 编程入门基础（五）：简单案例一

     解决了程序框架的问题， 我们就在 Eclipse 中开始写代码了。没有安装 Maven 的要参考前面的说明。

   一、构建

        我用的版本是这个，最新的版本

       


          File -->New --> Project .. --> Maven (Maven Project)  -->  Next --> Next , 在选择 Select an Archetype 中，catalog 选择 Internal，在下面选择



                   选择 Next --> Enter an artifact id， 我的 Group Id 还是我的姓氏，因为我不想使用我的公司域名， 然后在 Artifact id 上写上项目名称，这里命名为 teststorm，以后的案例都建在这下面即可。OK，finish ！



这就是我们构建的项目。当然我的JDK 是1.8的

二、开始编写代码

    1、修改 pom.xml

        打开 pom.xml 的代码，在 junit 的依赖下面加入以下内容：

       

<dependency>

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

  <artifactId>storm-core</artifactId>

  <version>1.1.0</version>

  <scope>provided</scope>

</dependency>



      2、先创建 Spout 类

      右键点击 域wangxn.teststorm创建 Spout 类，创建类的过程参考前面的内容。

        创建一个随机WordSpout
package wangxn.teststorm;

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

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

public class WordSpout extends BaseRichSpout {
private SpoutOutputCollector collector;
private static String[] words = {"星期一","星期二","星期三","星期四","星期五","星期六","星期日"};

public WordSpout() {
// TODO Auto-generated constructor stub

}

public void nextTuple() {
// TODO Auto-generated method stub
String word = words[new Random().nextInt(words.length)]; //随机取 words 字符串中一个词。
collector.emit(new Values(word));//发射元组到输出收集器

}

public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) {
// TODO Auto-generated method stub
this.collector=arg2;//定义数据源输出收集器

}

public void declareOutputFields(OutputFieldsDeclarer arg0) {
// TODO Auto-generated method stub
arg0.declare(new Fields("word"));//声明输出字段的名称为为 word

}

}


    3、再创建Bolt 类

      右键点击 域wangxn.teststorm创建 Bolt 类，创建类的过程参考前面的内容。

        创建一个ProcessBolt

package wangxn.teststorm;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

public class ProcessBolt extends BaseRichBolt {

public ProcessBolt() {
// TODO Auto-generated constructor stub
}

public void execute(Tuple arg0) {
// TODO Auto-generated method stub
//此处直接对接受到的元组进行处理，然后输出到控制台，这里没有将处理后的数据再送到输出收集器中。
String word = (String) arg0.getValue(0); //取得元组的数据
String out = "Hello ：" + word + "!";
System.err.println(out); //输出到控制台，使用 err.println 会显示红色，所以这里使用 err
}

public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
// TODO Auto-generated method stub

}

public void declareOutputFields(OutputFieldsDeclarer arg0) {
// TODO Auto-generated method stub

}

}


 4、再创建主类 WordStorm

   package wangxn.teststorm;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;

public class WordStorm {

public WordStorm() {
// TODO Auto-generated constructor stub
}

public static void main(String[] args) throws Exception {
// TODO Auto-generated method stub
//定义 TopologyBuilder
TopologyBuilder builder=new TopologyBuilder();
//定义 Spout
builder.setSpout("Spout_ID", new WordSpout());
//定义 Bolt
builder.setBolt("Bolt_ID", new ProcessBolt()).localOrShuffleGrouping("Spout_ID");

//下面开始定义运行模式
final Config config=new Config();
config.setDebug(false);

if (args != null && args.length > 0) {
//集群运行模式
config.setNumWorkers(3);

StormSubmitter.submitTopologyWithProgressBar(args[0], config, builder.createTopology());
}else {
//使用本地模式运行
final LocalCluster localCluster=new LocalCluster();

localCluster.submitTopology(WordStorm.class.getSimpleName(), config, builder.createTopology());
org.apache.storm.utils.Utils.sleep(90000);

localCluster.killTopology(WordStorm.class.getSimpleName());
localCluster.shutdown();
}

}

}


三、运行
    检查一下没有任何出错

  1、Eclipse 控制台运行

    工程名上点右键Run as --> Java Application，选择主类 WordStorm - wangxn.teststorm，点 OK，下面控制台开始输出。30秒后程序将中断，因为定义的延迟是30秒。输出的提示颜色是红的。

       


2、输出到服务器上运行

        右键点击工程名，选择 Export，--> Java (JAR file）  -->next ->选择 JARfile 的输出目录，此处输出到 home 目录，word storm.jar, 然后 finish即可。

        本地运行： $storm jar wordstorm.jar wangxn.teststorm.WordStorm

        集群运行：$storm jar wordstorm.jar wangxn.teststorm.WordStorm WordStorm



    也可以到 Ui web 浏览器上查看运行的 Topology