Spout 完成后终止 Storm 拓扑答案

【问题标题】：Killing Storm topology once Spout finishesSpout 完成后终止 Storm 拓扑
【发布时间】：2019-03-26 21:02:19
【问题描述】：

我创建了一个带有 Spout 的 Storm 拓扑，它发出许多元组用于基准测试。一旦从 spout 发出所有元组或拓扑中不再有任何元组流动，我想停止/终止我的拓扑。

这是我的拓扑结构。

LocalCluster cluster = new LocalCluster();
TopologyBuilder builder = new TopologyBuilder();
Config conf = new Config();
//Disabled ACK'ing for higher throughput
conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, 0); 

LoadGeneratorSource loadGenerator = new LoadGeneratorSource(runtime,numberOfTuplesToBeEmitted);
builder.setSpout("loadGenerator", loadGenerator);

//Some Bolts Here

while (loadGenerator.isRunning()){
//Active Waiting
}
//DO SOME STUFF WITH JAVA
cluster.killTopology("StormBenchmarkTopology");

问题是我在这个范围内引用的 loadGenerator 实例与在 spout 线程中运行的实例不同。因此，isRuning() 总是返回 true，即使在 spout 线程内，当没有更多元组要发出时，它的值为 false。

这是 LoadGeneratorSource 类的一部分。


public class LoadGeneratorSource extends BaseRichSpout {

    private final int throughput;
    private boolean running;
    private final long runtime;


    public LoadGeneratorSource(long runtime,int throughput) {
        this.throughput = throughput;
        this.runtime = runtime;
    }

    @Override
    public void nextTuple() {
        ThroughputStatistics.getInstance().pause(false);

        long endTime = System.currentTimeMillis() + runtime;
        while (running) {
            long startTs = System.currentTimeMillis();

            for (int i = 0; i < throughput; i++) {
                try {
                    emitValue(readNextTuple());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            while (System.currentTimeMillis() < startTs + 1000) {
                // active waiting
            }

            if (endTime <= System.currentTimeMillis())
                setRunning(false);
        }
    }

    public boolean isRunning() {
        return running;
    }

    public void setRunning(boolean running) {
        this.running = running;
    }

    //MORE STUFF

}

一旦不再有从 spout 发出或在拓扑中流动的元组，有人能告诉我停止我的拓扑的方法吗？提前感谢您的帮助。

【问题讨论】：

标签： apache-storm apache-storm-topology

【解决方案1】：

这似乎与Killing storm topology from spout 重复。请尝试那里给出的答案。

只是为了快速总结一下；您尝试执行此操作的方式不起作用，但您可以使用 spout 中的 NimbusClient 要求 Nimbus 终止您的拓扑。附带的好处是，一旦您部署到真正的集群，它也将起作用。

【讨论】：

我知道这个话题。我不想直接在 spout 中终止拓扑。元组完成后，我仍然需要在上述范围内处理一些事情。可能我不是很清楚。
我不清楚您的代码打算做什么。您是否只是在本地进行基准测试，而不打算将代码部署到真正的集群？如果是这样，只需将负载生成器放在静态字段中并从那里引用它（并使其成为线程安全的）。如果你想在真实的集群上使用你的代码，你需要重新考虑你在做什么。拆解代码需要发生在 spout 中，或者发生在使用“流结束”元组触发的 bolt 中。 “用 Java 做一些事情”代码不会在与真实集群上的 spout 相同的 JVM 中运行。
我明白你的意思。我最终可能会尝试使用集群，因此我认为我应该改变结构。谢谢你们的cmets。