共计 6084 个字符，预计需要花费 16 分钟才能阅读完成。
这篇文章主要介绍“storm 启动类怎么定义”，在日常操作中，相信很多人在 storm 启动类怎么定义问题上存在疑惑，丸趣 TV 小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”storm 启动类怎么定义”的疑惑有所帮助！接下来，请跟着丸趣 TV 小编一起来学习吧！
在 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 Map String, Object getComponentConfiguration() {
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() {
}
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 Map String, Object getComponentConfiguration() {
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 Map String, Object getComponentConfiguration() {
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 Map String, Object getComponentConfiguration() {
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 启动类怎么定义”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注丸趣 TV 网站，丸趣 TV 小编会继续努力为大家带来更多实用的文章！
正文完
