如何使用代码中的新螺栓更新现有的 Storm 拓扑？

Question

我正在编写一个使用 Apache Storm v1.1.2、Kafka v0.11.0.1、Zookeeper 3.4.6、Eureka 和 Cloud-Config 的 dockerized Java Spring 应用程序，所有这些都在由 Docker-Compose 编排的 Docker 容器中。

我通过 KafkaSpout 接收的元组有一个“值”字段，它是一个 protobuf 对象。我使用自定义反序列化器将我的对象从中取出进行处理。

我有一个基本的应用程序工作，我有一个螺栓，它打印传入的消息并根据 protobuf 对象中字段的值将它们路由到其他特定的螺栓。我还将 LocalCluster、Config 和 TopologyBuilder 作为 Spring Bean 工作。

目前我在 PostContruct 中设置了所有螺栓，但我需要能够动态添加螺栓，以根据 protobuf 对象的其他字段过滤传入消息并执行基本聚合函数（最大/最小/窗口平均值）。

我想使用 REST 控制器来执行此操作，但是如何在不丢失数据的情况下停止和启动拓扑？我也不希望从一开始就通过监听 Kafka 主题来重新启动拓扑，因为该系统将收到极高的负载。

这篇文章看起来很有希望，但我绝对希望整个过程自动化，所以我不会进入 Zookeeper https://community.hortonworks.com/articles/550/unofficial-storm-and-kafka-best-practices-guide.html

如何在代码中编辑现有拓扑以动态添加新螺栓？

Answer 1

你不能。 Storm 拓扑在提交后是静态的。如果您需要根据元组中的字段改变处理，最好的选择是预先提交您需要的所有螺栓。然后，您可以使用一个或多个检查元组的螺栓来改变元组通过拓扑的路径，并根据元组内容发送到特定的流。

例如做一个SplitterBolt

public void execute(Tuple input) {
  if (tuple.getIntegerByField("theDecider") == 1) {
    collector.emit("onlyOnes", tuple.getValues());
  } else {
    collector.emit("others", tuple.getValues());
  }
}

您在拓扑构建代码中的位置会有类似

builder.setSpout("kafka-spout", ...);
builder.setBolt("splitter", new SplitterBolt()).shuffleGrouping("kafka-spout");
builder.setBolt("countOnes", new CounterBolt()).shuffleGrouping("splitter", "onlyOnes");
builder.setBolt("countOthers", new CounterBolt()).shuffleGrouping("splitter", "others");