這篇文章主要介紹“storm啟動類怎么定義”，在日常操作中，相信很多人在storm啟動類怎么定義問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”storm啟動類怎么定義”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

10年積累的成都做網(wǎng)站、網(wǎng)站建設(shè)經(jīng)驗，可以快速應(yīng)對客戶對網(wǎng)站的新想法和需求。提供各種問題對應(yīng)的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡(luò)服務(wù)。我雖然不認識你，你也不認識我。但先做網(wǎng)站設(shè)計后付款的網(wǎng)站建設(shè)流程，更有鹿寨免費網(wǎng)站建設(shè)讓你可以放心的選擇與我們合作。

在storm集群中真正運行Topology的主要有三個實例:工作進程丶線程和任務(wù).

Storm集群中的每臺機器上都可以運行多個工作進程,每個工作進程又可以創(chuàng)建多個線程,每個線程可以執(zhí)行多個任務(wù).

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

一、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")對應(yīng)定義去接收，fiedsGrouping的new Fields("id")中的id要和componentId的對應(yīng)bolt中new Fields("id","message")中的"id"對應(yīng)就會以"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() {
    
  }


}

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 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啟動類怎么定義”的學習就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關(guān)知識，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>

名稱欄目：storm啟動類怎么定義
標題鏈接：http://chinadenli.net/article8/jhpdop.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供微信小程序、小程序開發(fā)、定制網(wǎng)站、用戶體驗、外貿(mào)建站、品牌網(wǎng)站設(shè)計

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)