示例简介
一个简单的Storm的示例，拓扑结构由一个Spout和一个Bolt组成，Spout负责从Redis中读取经纬度点（Coordinate）的数据，并且在这些点中随机挑选一个放入tuple中发射给Bolt。Bolt负责处理当前的点与上一个接收到的点之间的距离，并存储到Redis中的另一张表中。
Maven依赖
Demo使用的Storm版本为0.9.6，同时使用了Jedis来直接与Redis交互。在pom文件中添加如下依赖：
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>0.9.6</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.8.1</version>
</dependency>
创建Spout文件
Spout创建时的核心方法是 public void nextTuple()方法，Storm在这里将信息发送到collector的Stream中。
public void nextTuple() {
Utils.sleep(100);
Integer r = rand.nextInt(points.length);
String coor = points[r];
collector.emit(new Values(coor));
}
完整的代码如下：
package wech.storm_starter_demo;
import java.util.List;
import java.util.Map;
import java.util.Random;
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;
import backtype.storm.utils.Utils;
import redis.clients.jedis.Jedis;
public class PointReader implements IRichSpout {
/**
*
*/
private static final long serialVersionUID = 1L;
private SpoutOutputCollector collector;
private Random rand;
private Jedis jedis;
private String[] points;
private boolean complete = false;
private static String POINTS_KEY = "points";
public boolean isDistributed() {
return false;
}
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
this.rand = new Random();
this.jedis = new Jedis();
List<String>list = jedis.lrange(POINTS_KEY, 0, -1);
this.points = list.toArray(new String[1]);
}
public void ack(Object object) {
System.out.println("OK" + object);
}
public void activate() {
// TODO Auto-generated method stub
}
public void close() {
// TODO Auto-generated method stub
}
public void deactivate() {
// TODO Auto-generated method stub
}
public void fail(Object object) {
System.out.println("Fail" + object);
}
public void nextTuple() {
Utils.sleep(100);
Integer r = rand.nextInt(points.length);
String coor = points[r];
collector.emit(new Values(coor));
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("point"));
}
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
}
创建Bolt
bolt主要负责接收stream中的tuple信息并且进行处理，然后选择传递到下一个bolt或者通知Spout处理完成（失败）。代码如下：
package wech.storm_starter_demo;
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;
import backtype.storm.tuple.Values;
import redis.clients.jedis.Jedis;
public class PointHandle implements IRichBolt {
private OutputCollector collector;
private Jedis jedis;
private static String DISTANCE_KEY = "distancekey";
private String lastPoint;
private static Double R = 6378137.0;
public void cleanup() {
// TODO Auto-generated method stub
}
public Double[] splitPoint(String pointStr) {
String[] temp = pointStr.split(",");
Double[] fin = new Double[]{Double.parseDouble(temp[0]) , Double.parseDouble(temp[1])};
return fin;
}
public Double distance(Double lon1, Double lat1, Double lon2, Double lat2) {
Double a;
Double b;
lat1 = lat1 * Math.PI / 180.0;
lat2 = lat2 * Math.PI / 180.0;
a = lat1 - lat2;
b = (lon1 - lon2) * Math.PI / 180.0;
Double distance;
Double sa2,sb2;
sa2 = Math.sin(a / 2.0);
sb2 = Math.sin(b / 2.0);
distance = 2 * R * Math.asin(Math.sqrt(sa2 * sa2 + Math.cos(lat1)
* Math.cos(lat2) * sb2 * sb2));
return distance;
}
public void execute(Tuple input) {
String currentPoint = input.getString(0);
if (lastPoint.isEmpty()) {
lastPoint = currentPoint;
return;
} else {
Double[] lastCoor = this.splitPoint(lastPoint);
Double[] currentCoor = this.splitPoint(currentPoint);
Double distance = this.distance(lastCoor[0], lastCoor[1], currentCoor[0], currentCoor[1]);
String result = lastPoint + " -> " + currentPoint + " = " + distance;
System.out.println(result);
jedis.rpush(DISTANCE_KEY, result);
}
collector.ack(input);
}
public void prepare(Map arg0, TopologyContext arg1, OutputCollector collector) {
this.collector = collector;
jedis = new Jedis();
lastPoint = "";
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
}
创建Topology
创建一个Topology结构，将Spout和Bolt添加进去。代码如下：
package wech.storm_starter_demo;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
public class PointTopology {
public static void main(String[] args) throws InterruptedException, AlreadyAliveException, InvalidTopologyException {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new PointReader(),5);
builder.setBolt("bolt", new PointHandle(),8).shuffleGrouping("spout");
Config conf = new Config();
conf.setDebug(false);
if (args != null && args.length > 0) {
conf.setNumWorkers(3);
StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
}
else {
conf.setMaxTaskParallelism(3);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("point-handle", conf, builder.createTopology());
Thread.sleep(10000);
cluster.shutdown();
};
}
}
