Storm KafkaSpout如何知道所有的bolts都被执行了答案

【问题标题】：How does Storm KafkaSpout know all bolts are executedStorm KafkaSpout如何知道所有的bolts都被执行了
【发布时间】：2018-12-30 14:35:48
【问题描述】：

我的拓扑代码示例如下：

builder.setSpout("spout", new KafkaSpout);
builder.setBolt("bolt1", new Bolt1).shuffleGrouping("spout");
builder.setBolt("bolt2", new Bolt2).shuffleGrouping("bolt1");
builder.setBolt("bolt3", new Bolt3).shuffleGrouping("bolt2");

当bolt1 发出时，消息将被自动确认。但是当bolt2或bolt3出现异常时，无法重发这条消息，如何找回失败的消息？

【问题讨论】：

您能详细说明您的问题吗？我不清楚你在问什么。

标签： apache-kafka apache-storm kafka-consumer-api

【解决方案1】：

Storm 拥有 tuple trees 的概念。让我尝试使用问题中提供的示例进行解释。

当您的spout 调用collector.emit 方法时，新发出的元组，我们称之为tuple1，被添加到tuple tree。这个元组到达bolt1，因为它已经订阅了它，并将接收从spout 发出的数据。一旦它在execute 方法中接收到tuple1 作为输入，在处理输入后，一个新值将作为tuple2 发出，该值将添加到tuple1 之后的元组树中。在退出 execute 方法之前，通过隐式调用 collector.ack 来确认元组，这会告诉storm tuple1 已被处理，请将其从元组树中删除，现在保持tuple2 传递给bolt2 进行处理.

现在的问题是，如果bolt1 由于某种原因无法确认会发生什么。 Storm 会看到经过一段时间，也就是拓扑超时时间（默认为 30 秒）后，元组树还没有耗尽，因此它会从头开始重播元组，并遵循上述相同的过程。

希望我能够解释失败时会发生什么。更多详情请阅读this或观看this

【讨论】：