当前位置: 首页 > 知识库问答 >
问题:

Storm拓扑是否可以通过喷口或螺栓停用和重新激活?

冯鸿光
2023-03-14

当满足某些条件时,如遇到特殊类型的元组,我希望停用拓扑。这可以在喷口/螺栓中完成吗?如果是,是否也有办法从喷口/螺栓重新激活拓扑?

共有1个答案

和谦
2023-03-14

我在bolow代码中添加了所有三个动作来激活/停用/杀死。这可以从独立的java代码中调用(在喷口/螺栓外部)。

从喷口或螺栓停用是直接的,但重新激活将是棘手的,因为您的喷口/螺栓在停用后不会主动运行java程序

import backtype.storm.generated.KillOptions;
import backtype.storm.generated.Nimbus.Client;
import backtype.storm.utils.NimbusClient;
import backtype.storm.utils.Utils;

Client client = NimbusClient.getConfiguredClient(Utils.readStormConfig()).getClient();

client.activate(topologyName);
client.deactivate(topologyName);

KillOptions killOpts = new KillOptions();
killOpts.set_wait_secs(30);
client.killTopologyWithOpts(topologyName, killOpts);
 类似资料:
  • 因此,如果您使用基于JUnit的单元测试,是否建议您运行一个小型模拟拓扑(?)并测试该拓扑下的(或)的隐含契约?或者,是否可以使用JUnit,但这意味着我们必须仔细模拟Bolt的生命周期(创建它、调用、嘲弄等)?在这种情况下,被测类(螺栓/喷口)有哪些一般的测试点需要考虑? 其他开发人员在创建正确的单元测试方面做了什么? 我注意到有一个拓扑测试API(参见:https://github.com/x

  • 我对Apache Storm的性能有一个问题,主要是从喷口出来的。 我有一个从kestrel队列发出项目的拓扑。我获取大约2000个项目,每次在喷注中调用时,我都会发出一个。 我正在使用1个spout任务和1个spout执行器运行。我已将设置为10。 为什么每次调用之间有这么大的时间间隔?outputCollector在发出一个新元组之前是否正在等待听到每个元组的反馈? 我正在运行Java8和st

  • 我们有一个不想连续运行storm拓扑的用例。相反,有一组输入(10K+)应该在指定的时间被处理,Spout连续发射这些输入,并得到拓扑中其余螺栓的处理。处理完所有输入后,在我的喷注中就没有任何东西可以从nextTuple发出。 此时,我们希望拓扑进入Hibernate状态,并在每天晚上12:00重新启动进程。 在storm配置中是否有任何属性可以设置为每天运行一次拓扑并在处理完成后Hibernat

  • 在我的拓扑中,当元组从spout转移到bolt或从bolt转移到bolt时,我看到大约1-2 ms的延迟。我使用纳秒时间戳来计算延迟,因为整个拓扑运行在单个Worker中。拓扑是在集群中运行的,集群运行在具有生产能力的硬件中。 根据我的理解,在这种情况下,元组不需要序列化/反序列化,因为所有东西都在单个JVM中。我已经将大多数喷流和螺栓的并行性提示设置为5,并且喷流仅以每秒100的速率产生事件。我

  • 我对Apache Storm有一个奇怪的问题。我有一个Kafka连接到Kafka集群,里面有10条消息。 螺栓接收每条消息并正确处理,因为在Storm UI中,螺栓被列为“已确认”。然而,storm UI下面列出的喷口表示所有元组都失败了。 我相信这会导致喷口再次发出所有的信息。。。因此,我看到一个Storm螺栓打印出消息1-10,然后以相同的顺序一次又一次地打印出来。 我适当地调用了和方法,我只

  • 编辑:我向Bolt添加了一个。ack()(这要求我使用一个丰富的Bolt而不是基本的Bolt)并且遇到了同样的问题--没有任何信息告诉我Bolt正在处理元组。 如果有关系的话,我会在EC2实例上的CentOS映像上运行这个。如有任何帮助,不胜感激。 查看生成的Storm worker日志,我看到这一行: 下面几行如下: 工作日志的其余部分没有显示螺栓处理的消息的日志/打印。我不明白为什么螺栓似乎没