storm启动类怎么定义

122次阅读

共计 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 小编会继续努力为大家带来更多实用的文章！

正文完