（03）Storm编程案例

package demo;

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

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

//负责实时读取数据，然后发送给后续的bolt组件进行处理
public class DataSouceSpout extends BaseRichSpout {
    
    //模拟一些数据
    private String[] phones = {"iphone","xiaomi","moto","sunsumg","mate","huawei","nokia"};
    //得到上下文的信息
    private SpoutOutputCollector collector;
    
    //消息的处理方法
    public void nextTuple() {
        //模拟从外部读取数据
        Utils.sleep(500);
        //随机得到了一个手机的型号
        int index = new Random().nextInt(7);
        String phone = phones[index];
        //发送给后续的bolt
        this.collector.emit(new Values(phone));
    }

    //对这个spout进行初始化
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    //指定spout组件发送出去的数据的key
    public void declareOutputFields(OutputFieldsDeclarer declare) {
        declare.declare(new Fields("phone-name"));
    }
}

package demo;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

//将Spout发送过来的数据，转成大写
public class MyBoltA extends BaseBasicBolt {

    //业务的处理逻辑方法
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        //从spout组件中获取数据
        //方式一:
        //String phone = tuple.getStringByField("phone-name");
        //方式二
        String phone = tuple.getString(0);
        //处理数据
        String upperPhone = phone.toUpperCase();
        //将数据发送给下一个组件继续处理
        collector.emit(new Values(upperPhone));
    }

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

}

package demo;

import java.io.FileWriter;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

//将BoltA发送过来的数据，加上时间，并且写到文件中
public class MyBoltB extends BaseBasicBolt {

    private FileWriter fw = null;
    
    public void prepare(Map stormConf, TopologyContext context) {
        //对FileWriter进行初始化
        try {
            fw = new FileWriter("/usr/local/test/storm/mystormoutput.txt");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        //将boltA发送过来的数据，加上当前的时间
        //获取数据
        String upperPhoneName = tuple.getString(0);
        //业务逻辑
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String result = upperPhoneName +"   " + df.format(new Date());
        //将数据输出到文件系统中
        try {
            fw.write(result + "\n");
            fw.flush();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declare) {
        //这是最后一个bolt组件
    }

}

package demo;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;

//组装各个组件，并且提交任务到Storm集群
public class SubmitClient {

    public static void main(String[] args) throws Exception {
        //得到一个topology的构造器
        TopologyBuilder builder = new TopologyBuilder();
        //指定spout
        builder.setSpout("datasource-spout", new DataSouceSpout());
        //指定Bolt组件，还需要指定数据的来源
        builder.setBolt("boltA", new MyBoltA()).shuffleGrouping("datasource-spout");
        builder.setBolt("boltB", new MyBoltB()).shuffleGrouping("boltA");
        //生成一个具体的任务
        StormTopology phoneTopo = builder.createTopology();
        //指明任务的一些参数
        Config config = new Config();
        //希望storm集群分配6个worker来执行任务
        config.setNumWorkers(6);
        //提交任务
        StormSubmitter.submitTopology("mystormdemo", config, phoneTopo);
    }
}

[root@localhost apache-storm-0.9.2-incubating]# bin/storm jar /usr/local/test/storm/stormDemo.jar demo.SubmitClient

[root@localhost storm]# tail -f /usr/local/test/storm/mystormoutput.txt