storm啟動(dòng)類(lèi)怎么定義

176次閱讀

共計(jì) 6084 個(gè)字符，預(yù)計(jì)需要花費(fèi) 16 分鐘才能閱讀完成。

這篇文章主要介紹“storm 啟動(dòng)類(lèi)怎么定義”，在日常操作中，相信很多人在 storm 啟動(dòng)類(lèi)怎么定義問(wèn)題上存在疑惑，丸趣 TV 小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”storm 啟動(dòng)類(lèi)怎么定義”的疑惑有所幫助！接下來(lái)，請(qǐng)跟著丸趣 TV 小編一起來(lái)學(xué)習(xí)吧！

在 storm 集群中真正運(yùn)行 Topology 的主要有三個(gè)實(shí)例: 工作進(jìn)程丶線(xiàn)程和任務(wù).

Storm 集群中的每臺(tái)機(jī)器上都可以運(yùn)行多個(gè)工作進(jìn)程, 每個(gè)工作進(jìn)程又可以創(chuàng)建多個(gè)線(xiàn)程, 每個(gè)線(xiàn)程可以執(zhí)行多個(gè)任務(wù).

Storm 可靠性: 是通過(guò)對(duì)消息樹(shù)給定一個(gè)唯一的 ID, 每送一個(gè)消息, 都會(huì)同步發(fā)送一個(gè) ack 或 fail, 對(duì)于網(wǎng)絡(luò)的寬帶會(huì)有一定的消耗, 如果對(duì)于可靠性要求不高, 可以通過(guò)使用不同的 emit 接口關(guān)閉該模式.

一、storm 啟動(dòng)類(lèi)的定義。

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) 對(duì)應(yīng)定義去接收，fiedsGrouping 的 new Fields(id) 中的 id 要和 componentId 的對(duì)應(yīng) bolt 中 new Fields(id , message) 中的 id 對(duì)應(yīng)就會(huì)以 id 進(jìn)行分組

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() { }
}

到此，關(guān)于“storm 啟動(dòng)類(lèi)怎么定義”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注丸趣 TV 網(wǎng)站，丸趣 TV 小編會(huì)繼續(xù)努力為大家?guī)?lái)更多實(shí)用的文章！

正文完