（卡夫卡）暴风雨中返回的东西清单，不能将此清单传播到螺栓答案

【问题标题】：(Kafka) Spout on storm returning a list of things, can't spread this list to bolts（卡夫卡）暴风雨中返回的东西清单，不能将此清单传播到螺栓
【发布时间】：2015-12-02 22:14:10
【问题描述】：

我是 Storm 的新手，所以请多多包涵。

我有一项服务，每次我们与 结果列表通话时都会向 kafka 发送一条消息。

有一个 KafkaSpout 读取每条消息，这条消息包含前面提到的列表。

这只是一个 JSon，我可以打开它。现在，问题来了：

我在使用 jackson 的方案中执行此转换操作，但此方案基本上可以返回 Values 对象，它不是对象列表，而是基本上是字段值对列表。

另一件事可能是在 Bolt 中获取此 Values（它只是一个扩展的 ArrayList）对象，并将其展开为单个元素，并将它们中的每一个发送到下一个 Bolt。这是一个解决方案吗？我可以通过对 Bolt 的一次调用发出多个对象吗？

有更智能的解决方案吗？

【问题讨论】：

标签： java apache-kafka apache-storm

【解决方案1】：

是的，您可以从单个 Bolt 发出多个元组。查看位于here的示例bolt中的execute方法

public void execute(Tuple tuple) {
  String sentence = tuple.getString(0);
  for(String word: sentence.split(" ")) {
    _collector.emit(tuple, new Values(word));  //emits multiple tuples
  }
  _collector.ack(tuple);
}

如您所见，for 循环可以同时发出多个元组。这样做会创建一个更大的消息树。这可能会导致问题，具体取决于您的可靠性保证和数据大小。

根据我的经验，很难/不可能在 KafkaSpout 中操作数据。因此，请注意一些关于您的设置的说明。

我会尝试做的第一件事是更改服务发送到 Kafka 的内容。您可以将其发送为单个物品而不是一件大物品。如果您无法更改。
像您提到的那样设置多个 Bolt，Spout => UnwrapBolt => ProcessBolt，其中 UnwrapBolt 获取您的一个数据源并将它们作为单独的元组发出，然后 ProcessBolt 获取每个单独的元组并按照您的需要处理它们。

【讨论】：

谢谢你 morganw09dev。我实际上使用了第一种方法，我减少了消息传递，每条消息只有一个项目。一个问题：如果我对每个元组发出确认而不是在最后确认会发生什么？
我不知道这是否可能/可取？？？我的解释一直是，确认一个元组通常用于表示您已完成对特定 Bolt 的此元组的处理。因此，当您仍在同一个 Bolt 中处理元组时，确认元组没有多大意义。不过，我不知道这种观点有多正式。
哦，好的。我会试试的，我会让你知道会发生什么:)
期待会发生什么。
就结果而言，“emit-ack each tuple”与“acking at the end”有何不同？最后，execute方法总是每次处理一个Tuple（接收到），所以如果你发送ack，你就是在确认处理已经完成......如果你的意思是你想调用.ack方法在for 循环，好吧，我认为这是错误的，因为如果收集器无法将新元组发送到下一个 Bolt，那么您将永远不会再次获得相同的元组，因为它已经被确认了。此外，如果您使用事务拓扑，则不需要 ACK。