Apache Storm: 无法从多个bolts接收元组
创始人
2024-09-04 23:01:07
0

问题描述: 在Apache Storm中,有时候我们希望一个bolt能够从多个其他bolts接收元组,然后进行处理。但是默认情况下,每个bolt只能从一个上游bolt接收元组。

解决方法: 要解决这个问题,我们可以通过以下两种方式来实现一个bolt从多个bolts接收元组的功能。

  1. 使用多个输入流(Multiple input streams): Storm允许我们为bolt定义多个输入流,每个输入流都可以从一个独立的bolt接收元组。我们可以通过使用declareStream()方法在bolt的prepare()方法中声明多个输入流,并为每个输入流指定一个唯一的stream id。然后,在execute()方法中,我们可以使用getComponentId()方法获取bolt的组件id,并通过该id来判断当前接收到的元组是来自哪个输入流。

示例代码如下:

public class MyBolt extends BaseRichBolt {
  private OutputCollector collector;
  
  @Override
  public void prepare(Map topoConf, TopologyContext context, OutputCollector collector) {
    this.collector = collector;
    // 声明多个输入流
    // 第一个参数为输入流的id,第二个参数为源bolt的id,第三个参数为源bolt的输出字段列表
    context.declareStream("stream1", new Fields("field1"));
    context.declareStream("stream2", new Fields("field1"));
  }

  @Override
  public void execute(Tuple input) {
    String sourceComponent = input.getSourceComponent();
    String streamId = input.getSourceStreamId();
    
    if (streamId.equals("stream1")) {
      // 处理来自stream1的元组
    } else if (streamId.equals("stream2")) {
      // 处理来自stream2的元组
    }
    
    collector.ack(input);
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    // 声明输出字段列表
    declarer.declare(new Fields("outputField1", "outputField2"));
  }
}
  1. 使用自定义的数据结构传递元组(Passing tuples using custom data structure): 另一种方法是使用自定义的数据结构来传递元组。我们可以在接收到元组之后,将元组存储在一个数据结构中,然后在execute()方法中遍历该数据结构进行处理。

示例代码如下:

public class MyBolt extends BaseRichBolt {
  private OutputCollector collector;
  private List tuples;

  @Override
  public void prepare(Map topoConf, TopologyContext context, OutputCollector collector) {
    this.collector = collector;
    this.tuples = new ArrayList<>();
  }

  @Override
  public void execute(Tuple input) {
    // 将元组存储在列表中
    tuples.add(input);
    
    // 处理所有存储的元组
    for (Tuple tuple : tuples) {
      // 处理元组
    }
    
    collector.ack(input);
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    // 声明输出字段列表
    declarer.declare(new Fields("outputField1", "outputField2"));
  }
}

以上两种方法可以帮助我们实现一个bolt从多个bolts接收元组的功能。根据具体的需求和场景,选择合适的方法即可。

相关内容

热门资讯

科普常识(Wepoke新版)外... 科普常识(Wepoke新版)外挂透明挂辅助APP(软件)发牌机制(2025已更新)(哔哩哔哩);原来...
分享给玩家!边锋麻将有挂(we... 分享给玩家!边锋麻将有挂(wepoKE),太坑了原来是真的有挂(有挂透明)-哔哩哔哩,支持语音通讯、...
查到实测辅助!广东雀神辅助手机... 查到实测辅助!广东雀神辅助手机版本(透视)太坑了果真真的有挂(有挂技巧)-哔哩哔哩;广东雀神辅助手机...
透视科技!新版边锋掼蛋辅助器,... 1、透视科技!新版边锋掼蛋辅助器,德州辅助app官方最新版本介绍大全,详细有挂(有挂介绍)-哔哩哔哩...
总算了解(眯眯扑克)外挂透明挂... 大家肯定在之前眯眯扑克或者眯眯扑克中玩过总算了解(眯眯扑克)外挂透明挂辅助app(系统)发牌规律(2...
一分钟了解!边锋老友二打一有挂... 一分钟了解!边锋老友二打一有挂,wepoker私人局透视方法,详细有挂(有挂攻略)-哔哩哔哩是一款可...
实测分享!边锋麻将有挂(weP... 实测分享!边锋麻将有挂(wePOke),太坑了的确是真的有挂(有挂方法)-哔哩哔哩是一款可以让一直输...
一秒答解!广东雀神挂件去哪买(... 相信很多朋友都在电脑上玩过广东雀神挂件去哪买吧,但是很多朋友都在抱怨用电脑玩起来不方便。为此小编给大...
必备攻略(wpk一直输)外挂透... 1、必备攻略(wpk一直输)外挂透明挂辅助APP(线上)德州ai机器人(2025已更新)(哔哩哔哩)...
分享个大家!边锋游戏辅助器,x... 分享个大家!边锋游戏辅助器,xpoker辅助器,详细有挂(有挂总结)-哔哩哔哩1、点击下载安装,微扑...