前段时间,项目中使用到了流式处理方面的技术,学习了一下storm,编写了一个小实例。
1.引入jar包
4.0.0 org.springframework.boot spring-boot-starter-parent2.3.12.RELEASE com.example control0.0.1-SNAPSHOT control Demo project for Spring Boot 1.8 org.springframework.boot spring-boot-starterlog4j-to-slf4j org.apache.logging.log4j jul-to-slf4j org.slf4j logback-classic ch.qos.logback org.springframework.boot spring-boot-starter-weborg.apache.storm storm-core2.2.1 com.codahale.metrics metrics-core3.0.2 slf4j-api org.slf4j org.springframework.boot spring-boot-maven-pluginorg.projectlombok lombokorg.apache.maven.plugins maven-shade-pluginpackage shade false commons-logging:commons-logging javax.servlet:servlet-api javax.mail:javax.mail-api
2.编写程序:
(1)编写输入流类 package test.storm; import java.util.Map; import java.util.Random; import org.apache.storm.spout.SpoutOutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.topology.base.baseRichSpout; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Values; public class WordSpout extends baseRichSpout { private SpoutOutputCollector collector; private static String[] words = {"星期一","星期二","星期三","星期四","星期五","星期六","星期日"}; public WordSpout() { System.out.println("--====================WordSpout===---------------"); } public void nextTuple() { //随机取 words 字符串中一个词。 String word = words[new Random().nextInt(words.length)]; //发射元组到输出收集器 collector.emit(new Values(word)); } public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) { this.collector=arg2; //定义数据源输出收集器 } public void declareOutputFields(OutputFieldsDeclarer arg0) { // TODO Auto-generated method stub //声明输出字段的名称为为 word arg0.declare(new Fields("word")); } }
(2)整理数据流并输出
package test.storm;
import java.util.Map;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.baseRichBolt;
import org.apache.storm.tuple.Tuple;
public class ProcessBolt extends baseRichBolt {
public ProcessBolt() {
// TODO Auto-generated constructor stub
System.out.println("-----------ProcessBolt==============");
}
@Override
public void execute(Tuple arg0) {
// TODO Auto-generated method stub
//此处直接对接受到的元组进行处理,然后输出到控制台,这里没有将处理后的数据再送到输出收集器中。
//取得元组的数据
String word = (String) arg0.getValue(0);
String out = "Hello :" + word + "!";
//输出到控制台,使用 err.println 会显示红色,所以这里使用 err
System.err.println(out);
}
@Override
public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
// TODO Auto-generated method stub
System.out.println("-----------prepare==============");
}
@Override
public void declareOutputFields(OutputFieldsDeclarer arg0) {
// TODO Auto-generated method stub
System.out.println("-----------declareOutputFields==============");
}
}
(3)编写启动类:
package test.storm; import org.apache.storm.Config; import org.apache.storm.LocalCluster; import org.apache.storm.StormSubmitter; import org.apache.storm.topology.TopologyBuilder; public class WordStorm { public WordStorm() { // TODO Auto-generated constructor stub } public static void main(String[] args) throws Exception { // TODO Auto-generated method stub //定义 TopologyBuilder TopologyBuilder builder=new TopologyBuilder(); //定义 Spout builder.setSpout("Spout_ID", new WordSpout()); //定义 Bolt builder.setBolt("Bolt_ID", new ProcessBolt()).localOrShuffleGrouping("Spout_ID"); //下面开始定义运行模式 final Config config=new Config(); config.setDebug(true); //设置workers config.setNumWorkers(1); config.setMaxSpoutPending(1); if (args != null && args.length > 0) { //集群运行模式 config.setNumWorkers(3); StormSubmitter.submitTopologyWithProgressBar(args[0], config, builder.createTopology()); }else { //使用本地模式运行 final LocalCluster localCluster=new LocalCluster(); localCluster.submitTopology(WordStorm.class.getSimpleName(), config, builder.createTopology()); org.apache.storm.utils.Utils.sleep(20000); localCluster.killTopology(WordStorm.class.getSimpleName()); localCluster.shutdown(); } } }
运行结果:
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)