Storm-Jedis-Redis自制测试程序(仅作分享与记录）

1         建立JAVA工程

       在Java Build Path的Libraries中，添加外部JARS，将commons-collections，twitter/lib和storm/lib中的JAR包都加入工程，并添加storm之外的JAR包（Jedis）。

 


2         建立主类（Randomwords）

       在该类中，主要需要编写main(String[]args) 函数，在其中建立Topology。需要注意 args为程序在调用时所输入的参数，这里只是Topology的name，故在运行jar包时，最后别忘了给我们的拓扑结构起个名字，这样就能在Storm UI上看到了。

Randomwords.java代码如下：

packagestorm;

importbacktype.storm.Config;
importbacktype.storm.LocalCluster;
importbacktype.storm.StormSubmitter;
importbacktype.storm.topology.TopologyBuilder;

publicclass Randomwords {

public static void main(String[] args) {
// TODO Auto-generated method stub
//实例化TopologyBuilder类
TopologyBuilder myBuilder = newTopologyBuilder();
//设置喷发节点并分配并发数
myBuilder.setSpout("testSpout",new Spout(), 2);
//设置数据处理节点并分配并发数，链接Bolt和Spout
myBuilder.setBolt("testBolt",new Bolt(), 3).shuffleGrouping(
"testSpout");
//设置参数
Config config = new Config();
config.setDebug(true);
if (args != null &&args.length > 0) {
config.setNumWorkers(1);//并发工作个数
try {
StormSubmitter.submitTopology(args[0],config,
myBuilder.createTopology());
} catch (Exception e) {
// TODOAuto-generated catch block by StormSubmitter
e.printStackTrace();
}
} else {
//本地模式运行的启动代码
config.setMaxTaskParallelism(1);
LocalCluster myCluster =new LocalCluster();
myCluster.submitTopology("LocalTest",config,
myBuilder.createTopology());
}
}

}
3         建立Spout类（Spout）

       该类的作用是通过编写 nextTuple() 函数来获取数据流。必须编写的方法会在给该类添加“extendsBaseRichSpout ”时提示自动生成。

Spout.java代码如下：

packagestorm;

importjava.util.Map;
importjava.util.Random;

importbacktype.storm.spout.SpoutOutputCollector;
importbacktype.storm.task.TopologyContext;
importbacktype.storm.topology.OutputFieldsDeclarer;
importbacktype.storm.topology.base.BaseRichSpout;
importbacktype.storm.tuple.Fields;
importbacktype.storm.tuple.Values;

@SuppressWarnings("serial")
publicclass Spout extends BaseRichSpout {
//设置该测试程序的数据来源
private static String[] info = newString[] {
"Until tonight.Something was different tonight. ",
"There was an edge tothis darkness that made his hackles rise. ",
"Nine days they hadbeen riding, "
+ "northand northwest and then north again, "
+"farther and farther from the Wall, "
+ "hardon the track of a band of wildling raiders.",
" Each day had beenworse than the day that had come before it.",
" Today was the worstof all. ",
"A cold wind wasblowing out of the north, "
+ "andit made the trees rustle like living things. ",
"All day, Will hadfelt as though something were watching him, "
+"something cold and implacable that loved him not.",
" Gared had felt ittoo. ",
"Will wanted nothingso much as to ride hellbent for the safety of the Wall,"
+ " butthat was not a feeling to share with your commander." };

private Random random = new Random();

private SpoutOutputCollector collector;

//初始化collector
@SuppressWarnings("rawtypes")
@Override
public void open(Map arg0,TopologyContext arg1, SpoutOutputCollector arg2) {
// TODO Auto-generated method stub
this.collector = arg2;
}

//在SpoutTracker类中被调用，每调用一次就可以向storm集群中发射一条数据（一个tuple元组），该方法会被不停的调用
@Override
public void nextTuple() {
// TODO Auto-generated method stub
try {
int randomindex =random.nextInt(9);
String msg = randomindex +": " + info[randomindex];
collector.emit(newValues(msg));
Thread.sleep(200);
} catch (InterruptedException e) {
// TODO Auto-generatedcatch block
e.printStackTrace();
}
}

//定义字段id，该id在简单模式下没有用处，但在按照字段分组的模式下有很大的用处。
@Override
public voiddeclareOutputFields(OutputFieldsDeclarer arg0) {
// TODO Auto-generated method stub
arg0.declare(newFields("source"));
}

}

4         建立Bolt类（Bolt）

       该类实现数据的处理，即编辑execute(Tuple arg0, BasicOutputCollector arg1)方法。由于当前Redis版本不支持集群，暂时只是进行分片存储。信息存入不同的机器中，通过ShardedJedis合成一整个内存存储系统，单独用本地客户端看其中的一台机器，只能看到部分的存储内容，这里暂时只用一台机器进行存储。

Bolt.java代码如下：

package storm;

import java.util.Arrays;
import java.util.List;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
importbacktype.storm.topology.BasicOutputCollector;
importbacktype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

@SuppressWarnings("serial")
public classBolt extends BaseBasicBolt {

privateint testCount = 0;
@SuppressWarnings("resource")
@Override
publicvoid execute(Tuple arg0,BasicOutputCollector arg1) {
if (testCount >= 18) {
return;
}
// TODO Auto-generated method stub
String msg = arg0.getString(0);
if (msg != null) {

String msgList[] =msg.split(":");
String key =msgList[0].trim();
String value = msgList[1];
if (Integer.parseInt(key) < 0)
return;
// 客户端信息
List<JedisShardInfo>shards = Arrays.asList(
newJedisShardInfo("192.168.1.122", 6379)
);
// 切片客户端连接
ShardedJedis shardedJedis =new ShardedJedis(shards);
String result =shardedJedis.set(key, value);
System.out.println(key + ":" +result);
shardedJedis.disconnect();
testCount++;
}
}

@Override
publicvoiddeclareOutputFields(OutputFieldsDeclarer arg0) {
// TODO Auto-generated method stub
arg0.declare(new Fields("info"));
}

}

 
5         导出成jar包

       需要用MANIFEST.MF文件来配置程序运行的部分参数：

Manifest-Version:1.0
Main-Class:storm.Randomwords
Class-Path: /usr/local/lib/Jedis.2.6.0.jar
 

注意
4000
：

1)       最后一行的后面必须多出一行，否则最后一行设置将无效；

2)       在打包时，NEXT到最后一步时，选择“使用工程中已有的manifest”，选择上述文件；

3)       生成的jar包中就有/META-INFO/MANIFEST.MF了，且内容同上述文件一致；

4)       每个参数后的“：”后都必须有“空格”；

5)       Class-Path:内的jar包用“空格”或“回车+空格*2”来隔开；

6)       引用的Storm之外的jar包目录为绝对目录，需要保证storm的每台计算机的对应目录下有这些jar包。