这篇文章主要介绍“storm启动类怎么定义”,在日常操作中,相信很多人在storm启动类怎么定义问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”storm启动类怎么定义”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
10年积累的成都做网站、网站建设经验,可以快速应对客户对网站的新想法和需求。提供各种问题对应的解决方案。让选择我们的客户得到更好、更有力的网络服务。我虽然不认识你,你也不认识我。但先做网站设计后付款的网站建设流程,更有鹿寨免费网站建设让你可以放心的选择与我们合作。
在storm集群中真正运行Topology的主要有三个实例:工作进程丶线程和任务.
Storm集群中的每台机器上都可以运行多个工作进程,每个工作进程又可以创建多个线程,每个线程可以执行多个任务.
Storm可靠性:是通过对消息树给定一个唯一的ID,每送一个消息,都会同步发送一个ack或fail,对于网络的宽带会有一定的消耗,如果对于可靠性要求不高,可以通过使用不同的emit接口关闭该模式.
一、storm启动类的定义。
package com.cmsz.storm.trading.test; import backtype.storm.topology.TopologyBuilder; import backtype.storm.tuple.Fields; public class MainStorm { public static void main(String[] args) throws Exception { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("A", new ASpout()); builder.setBolt("B", new BBolt()).shuffleGrouping("A", "streamId_B");//componentId和streamId builder.setBolt("C", new CBolt()).shuffleGrouping("A", "streamId_C");//componentId和streamId builder.setBolt("D", new DBolt()).fieldsGrouping("B", new Fields("id"));//componentId和streamId builder.setBolt("E", new EBolt()).fieldsGrouping("C", new Fields("id")); Config conf = new Config(); if (args != null && args.length > 0) { conf.setNumWorkers(1); StormSubmitter.submitTopology(args[0], conf, builder.createTopology()); } else { LocalCluster cluster = new LocalCluster(); cluster.submitTopology("myTopo", conf, builder.createTopology()); } } }
二、spout定义了streamId,接受的bolt要定义componentId与spout中定义的streamId("streamId_B"、"streamId_C")对应定义去接收,fiedsGrouping的new Fields("id")中的id要和componentId的对应bolt中new Fields("id","message")中的"id"对应就会以"id"进行分组
package com.cmsz.storm.trading.test; import java.util.Map; import java.util.Random; import backtype.storm.spout.SpoutOutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseRichSpout; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Values; import backtype.storm.utils.Utils; public class ASpout extends BaseRichSpout{ SpoutOutputCollector collector; @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { this.collector = collector; } @Override public void nextTuple() { Utils.sleep(10); final String[] words = new String[] {"B_nathan", "C_mike", "B_jackson", "C_golda", "B_bertels"}; final Random rand = new Random(); final String word = words[rand.nextInt(words.length)]; if(word.indexOf("B_")>-1){ collector.emit("streamId_B",new Values(word)); }else if(word.indexOf("C_")>-1){ collector.emit("streamId_C",new Values(word)); } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream("streamId_B", new Fields("streamId_B")); declarer.declareStream("streamId_C", new Fields("streamId_C")); } @Override public void ack(Object msgId) { super.ack(msgId); } @Override public void fail(Object msgId) { super.fail(msgId); } }
package com.cmsz.storm.trading.test; import java.util.Map; import backtype.storm.task.TopologyContext; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.IBasicBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; public class BBolt implements IBasicBolt { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id","message")); } @Override public MapgetComponentConfiguration() { return null; } @Override public void prepare(Map stormConf, TopologyContext context) { } @Override public void execute(Tuple input, BasicOutputCollector collector) { String msg = input.getString(0); System.out.println(msg); collector.emit(new Values(msg,msg+"BBolt")); } @Override public void cleanup() { } }
package com.cmsz.storm.trading.test; import java.util.Map; import backtype.storm.task.TopologyContext; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.IBasicBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; public class CBolt implements IBasicBolt { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id","message")); } @Override public MapgetComponentConfiguration() { return null; } @Override public void prepare(Map stormConf, TopologyContext context) { } @Override public void execute(Tuple input, BasicOutputCollector collector) { String msg = input.getString(0); System.out.println(msg); collector.emit(new Values(msg,msg+"CBolt")); } @Override public void cleanup() { } }
package com.cmsz.storm.trading.test; import java.util.Map; import backtype.storm.task.TopologyContext; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.IBasicBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; public class DBolt implements IBasicBolt { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("message")); } @Override public MapgetComponentConfiguration() { return null; } @Override public void prepare(Map stormConf, TopologyContext context) { } @Override public void execute(Tuple input, BasicOutputCollector collector) { System.out.println("DBolt"+input.getString(0)); } @Override public void cleanup() { } }
package com.cmsz.storm.trading.test; import java.util.Map; import backtype.storm.task.TopologyContext; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.IBasicBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; public class EBolt implements IBasicBolt { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("message")); } @Override public MapgetComponentConfiguration() { return null; } @Override public void prepare(Map stormConf, TopologyContext context) { } @Override public void execute(Tuple input, BasicOutputCollector collector) { System.out.println("EBolt"+input.getString(0)); } @Override public void cleanup() { } }
到此,关于“storm启动类怎么定义”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!