Stormでメッセージのフィルタリングとルーティング機能を実装する方法は何ですか？

Stormには、Boltを定義することで、メッセージのフィルタリングやルーティング機能を実現することができます。具体的な手順は次の通りです：

1. 特定条件を満たしたメッセージを処理するためのフィルターBoltを作成します。このBoltでは、メッセージの内容や特定の条件に基づいて、メッセージを処理する必要があるかどうかを判断することができます。必要な場合はメッセージを継続的に送信でき、不要な場合はメッセージを無視することができます。

public class FilterBolt extends BaseRichBolt {
  private OutputCollector collector;

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

  @Override
  public void execute(Tuple input) {
    // 进行消息过滤逻辑
    if (/* 判断条件 */) {
      // 继续传递消息
      collector.emit(input, new Values(/* 消息内容 */));
    } else {
      // 忽略该消息
      collector.ack(input);
    }
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declare(new Fields("filteredMessage"));
  }
}

1. メッセージルーティング機能を処理するために、Boltルーターを作成します。このBoltでは、メッセージの内容や特定の条件に基づいて、どの目標Boltにメッセージをルーティングするかを決定できます。

public class RouterBolt extends BaseRichBolt {
  private OutputCollector collector;

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

  @Override
  public void execute(Tuple input) {
    // 进行消息路由逻辑
    if (/* 判断条件 */) {
      // 路由到目标Bolt中
      collector.emit("targetBolt", input, new Values(/* 消息内容 */));
    } else {
      // 路由到其他Bolt中
      collector.emit("otherBolt", input, new Values(/* 消息内容 */));
    }

    collector.ack(input);
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declareStream("targetBolt", new Fields("routedMessage"));
    declarer.declareStream("otherBolt", new Fields("routedMessage"));
  }
}

1. Topology 内でフィルターとルーターのBolt を設定し、TopologyBuilder でメッセージのフローのパスを指定します。

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("spout", new MySpout(), 1);
builder.setBolt("filterBolt", new FilterBolt(), 2).shuffleGrouping("spout");
builder.setBolt("routerBolt", new RouterBolt(), 2).shuffleGrouping("filterBolt");

Config conf = new Config();
conf.setDebug(true);

LocalCluster cluster = new LocalCluster();
cluster.submitTopology("myTopology", conf, builder.createTopology());

上記の手順を経て、Stormでメッセージのフィルタリングおよびルーティング機能を実現することができます。必要に応じて、より複雑なメッセージ処理ロジックを実装するためにBoltをカスタマイズおよび拡張することができます。