apache-storm - 当我使用 storm 时，如何确保具有多个输入的 bolt 仅在所有输入到达时才处理？

Question

拓扑结构如下所示。

如何确保具有多个输入的 bolt 仅在所有输入到达时才处理？

Answer 1

Bolt.execute()为每个传入的元组调用，不管生产者是什么(你不能改变这个)。如果您想一次处理来自不同生产者的多个元组，则需要编写自定义 UDF 代码。

1. 每个生产者都需要一个输入缓冲区，它可以缓冲传入的元组(可能是 LinkedList<Tuple> 作为 bolt 成员)
2. 对于每个传入的元组，您将元组添加到相应的缓冲区(您可以访问元组元数据中的生产者信息，通过。input.getSourceComponent()
3. 将元组添加到缓冲区后，检查每个缓冲区是否至少包含一个元组:如果是，则从每个缓冲区中取出一个元组并对其进行处理(处理后，再次检查缓冲区，直到至少有一次缓冲区为空) -- 不，只是返回并且不处理任何东西。