1 使用Eclipse创建maven工程
2 pom.xml内增加storm依赖:
<dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <version>0.9.3</version> </dependency>
3 自定义一个 spout 一个bolt, 在spout内不断产生i 并发送到bolt内, 在bolt内实现累加并打印效果
此案例主要是结合storm简介 中关于storm术语(车头 轨道 车次 车厢 车厢内人数tuple)
对应代码写法的一个参考而已
4 代码如下:
package changping.houzhihoujue.storm; import java.util.Map; import java.util.concurrent.TimeUnit; import ch.qos.logback.core.util.TimeUtil; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.spout.SpoutOutputCollector; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.TopologyBuilder; import backtype.storm.topology.base.BaseRichBolt; import backtype.storm.topology.base.BaseRichSpout; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; /** * 本地运行: * 实现累加 * @author zm * */ public class MyLocalStormTopology { /** * 组装火车 轨道 并让火车在轨道上行驶 * @throws InterruptedException */ public static void main(String[] args) throws InterruptedException { // 祖品列车 TopologyBuilder topologyBuilder = new TopologyBuilder(); topologyBuilder.setSpout("1", new MySpout2()); // 定义1号车厢 topologyBuilder.setBolt("2", new MyBolt1()).shuffleGrouping("1");// 定义2号车厢 并和1号车厢连接起来 // 造出轨道 LocalCluster localCluster = new LocalCluster();// 造出轨道 在本地运行 Config config = new Config(); // 轨道上运行列车, 三个参数分别为:定义的列车名,列车服务人员,轨道上跑的列车本身 localCluster.submitTopology(MyLocalStormTopology.class.getSimpleName(), config, topologyBuilder.createTopology()); TimeUnit.SECONDS.sleep(99999);// 设置列车运行时间 localCluster.shutdown();// 跑完后就停止下来, 否则storm是永不停止 } } //创建火车头 class MySpout2 extends BaseRichSpout { private Map conf; private TopologyContext context; private SpoutOutputCollector collector; // 此方法首先被调用 打开storm系统外的数据源 public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { this.conf = conf; this.context = context; this.collector = collector; } private int i = 0; // 认为是NameNode的heartbeat,永无休息的死循环的调用 并是线程安全的操作, 这里每一次调用此方法 将i++发送到bolt public void nextTuple() { System.err.println(i); // 将数据(i++)放在弹壳(Values)中,并发送给bolt this.collector.emit(new Values(i++)); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } } //声明输出的字段的名称为 v1 只有在输出给别人时才会重写此方法 public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("v1")); } } // 创建车厢 class MyBolt1 extends BaseRichBolt{ private Map stormConf; private TopologyContext context; private OutputCollector collector; // 准备下要对接收storm spout发送来的数据 public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) { this.stormConf = stormConf; this.context = context; this.collector = collector; } private int sum = 0; // 死循环,用于接收bolt送来的数据 这里storm每调用一次此方法 则获取发送来的tuple数据 public void execute(Tuple input) { int i = input.getIntegerByField("v1"); sum += i; System.err.println(sum); } // 只有向外发送数据时 此方法才会被调用 否则 不要实现此方法 public void declareOutputFields(OutputFieldsDeclarer declarer) { } }