Storm初探

Storm是一种分布式流式处理技术，这一点和Spark与Hadoop的批处理特性有明显的区别。

在数据连续产生，响应时延要求较低的场景中，Storm具有Spark不可比拟的优势。

网络性能监控系统中，Storm可以在秒级Dashboard监控，分钟级告警监控中大显生手。

学习任何技术，首先从Hello Wold开始，Storm也不例外，下面代码实现了这样一个例子：

（1） NamesReader Spout读取一行名字字符串，发送给NameSpliter；

（2） NameSpliter Bolt按照空格分割名字字符串，每个名字发送给HelloWorld；

（3） HelloWorld Bolt打印Hello world + 名字。

注：Spout是Storm有向网络的起始节点，Bolt是Storm有向网络的其他节点。数据在Storm有向网络中流动，节点则可以对流经的数据进行处理。

1、 名字字符串读取Spout

package com.coshaho.learn.storm;

import java.util.ArrayList;

import java.util.List;

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;

/**

 *

 * NamesReaderSpout.java Create on 2017年6月4日 下午10:57:32

 *

 * 类功能说明:   读取名字列表并派发

 *

 * Copyright: Copyright(c) 2013

 * Company: COSHAHO

 * @Version 1.0

 * @Author coshaho

 */

public class NamesReaderSpout implements IRichSpout

{

    private static final long serialVersionUID = 1L;

    private SpoutOutputCollector collector;

    @SuppressWarnings("rawtypes")

    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector)

    {

        this.collector = collector;

    }

    public void nextTuple()

    {

        List<Object> list = new ArrayList<Object>();

        try

        {

            Thread.sleep(5 * 1000);

        }

        catch (InterruptedException e)

        {

            e.printStackTrace();

        }

        System.out.println();

        list.add("刘备 关羽 张飞");

        // 第一个参数是传递的业务数据，第二个参数是消息标识，用于追踪消息是否正确处理

        this.collector.emit(list, "stream");

        list.clear();

        list.add("曹操 郭嘉 荀彧");

        this.collector.emit(list, "stream");

    }

    public void declareOutputFields(OutputFieldsDeclarer declarer)

    {

        // 必须设置，否则topo启动失败，names对应传递消息第一个元素，即list(0)

        declarer.declare(new Fields("names"));

    }

    public void close() {

        // TODO Auto-generated method stub

    }

    public void activate() {

        // TODO Auto-generated method stub

    }

    public void deactivate() {

        // TODO Auto-generated method stub

    }

    public void ack(Object msgId) {

        // TODO Auto-generated method stub

    }

    public void fail(Object msgId) {

        // TODO Auto-generated method stub

    }

    public Map<String, Object> getComponentConfiguration() {

        // TODO Auto-generated method stub

        return null;

    }

}

2、 名字字符串分割Bolt

package com.coshaho.learn.storm;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import backtype.storm.task.OutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.IRichBolt;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Tuple;

/**

 *

 * NamesSpliterBolt.java Create on 2017年6月4日 下午10:58:08

 *

 * 类功能说明:   名字列表按空格分割

 *

 * Copyright: Copyright(c) 2013

 * Company: COSHAHO

 * @Version 1.0

 * @Author coshaho

 */

public class NamesSpliterBolt implements IRichBolt

{

    private static final long serialVersionUID = 1L;

    private OutputCollector collector;

    @SuppressWarnings("rawtypes")

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)

    {

        this.collector = collector;

    }

    public void execute(Tuple input)

    {

        // 打印线程号用于追踪Storm的分配策略

        Thread current = Thread.currentThread();

        String names = input.getString(0);

        System.out.println("准备拆分" + names + "。当前线程号是" + current.getId() + "。");

        List<Tuple> inputList = new ArrayList<Tuple>();

        inputList.add(input);

        String[] nameArray = names.split(" ");

        for(String name : nameArray)

        {

            List<Object> splitList = new ArrayList<Object>();

            splitList.add(name);

            collector.emit(inputList, splitList);

        }

        collector.ack(input);

    }

    public void declareOutputFields(OutputFieldsDeclarer declarer)

    {

        // 必须设置，否则topo启动失败

        declarer.declare(new Fields("name"));

    }

    public void cleanup() {

        // TODO Auto-generated method stub

    }

    public Map<String, Object> getComponentConfiguration() {

        // TODO Auto-generated method stub

        return null;

    }

}

3、 HelloWorld Bolt

package com.coshaho.learn.storm;

import java.util.Map;

import backtype.storm.task.OutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.IRichBolt;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.tuple.Tuple;

/**

 *

 * HelloWorldBolt.java Create on 2017年6月4日 下午10:58:26

 *

 * 类功能说明:   Storm Hello World

 *

 * Copyright: Copyright(c) 2013

 * Company: COSHAHO

 * @Version 1.0

 * @Author coshaho

 */

public class HelloWorldBolt implements IRichBolt

{

    private static final long serialVersionUID = 1L;

    private OutputCollector collector;

    @SuppressWarnings("rawtypes")

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)

    {

        this.collector = collector;

    }

    public void execute(Tuple input)

    {

        // 打印线程号用于追踪Storm的分配策略

        Thread current = Thread.currentThread();

        String name = input.getString(0);

        System.out.println("你好，" + name + "。欢迎来到Storm世界。当前线程号是" + current.getId() + "。");

        collector.ack(input);

    }

    public void cleanup() {

        // TODO Auto-generated method stub

    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {

        // TODO Auto-generated method stub

    }

    public Map<String, Object> getComponentConfiguration() {

        // TODO Auto-generated method stub

        return null;

    }

}

4、 Storm TOPO网络任务启动

package com.coshaho.learn.storm;

import backtype.storm.Config;

import backtype.storm.LocalCluster;

import backtype.storm.topology.TopologyBuilder;

import backtype.storm.tuple.Fields;

public class StormTest

{

    public static void main(String[] args) throws InterruptedException

    {

        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("names-reader", new NamesReaderSpout());

        // 启动两个名字分割Task，名字列表随机分配给一个Task

        builder.setBolt("names-spliter", new NamesSpliterBolt(), 2)

            .shuffleGrouping("names-reader");

        // 启动两个Hello World Task，相同名字发送到同一个Task

        builder.setBolt("hello-world", new HelloWorldBolt(), 2)

            .fieldsGrouping("names-spliter", new Fields("name"));

        Config conf = new Config();

        LocalCluster cluster = new LocalCluster();

        cluster.submitTopology("storm-test", conf, builder.createTopology());

    }

}

5、 执行结果

准备拆分刘备 关羽 张飞。当前线程号是85。

你好，刘备。欢迎来到Storm世界。当前线程号是79。

你好，关羽。欢迎来到Storm世界。当前线程号是81。

你好，张飞。欢迎来到Storm世界。当前线程号是81。

准备拆分曹操 郭嘉 荀彧。当前线程号是87。

你好，荀彧。欢迎来到Storm世界。当前线程号是79。

你好，曹操。欢迎来到Storm世界。当前线程号是81。

你好，郭嘉。欢迎来到Storm世界。当前线程号是81。

准备拆分刘备 关羽 张飞。当前线程号是87。

准备拆分曹操 郭嘉 荀彧。当前线程号是85。

你好，荀彧。欢迎来到Storm世界。当前线程号是79。

你好，曹操。欢迎来到Storm世界。当前线程号是81。

你好，郭嘉。欢迎来到Storm世界。当前线程号是81。

你好，刘备。欢迎来到Storm世界。当前线程号是79。

你好，关羽。欢迎来到Storm世界。当前线程号是81。

你好，张飞。欢迎来到Storm世界。当前线程号是81。

6、 maven依赖

<dependency>

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

    <artifactId>storm-core</artifactId>

    <version>0.9.2-incubating</version>

</dependency>