风暴中的延迟队列/消息处理

Question

在我的风暴拓扑中，在处理流时，我希望将一些消息的处理推迟到将来的某个时间点。这样做有哪些合理的选择？

到目前为止，我考虑了以下几点：

• 使用Java的Thread.sleep。(然而，根据一些讨论，这并不是有效利用Storm资源的推荐方法。)
• 使用延迟队列..。
  • 特别是，尝试java.util.concurrent.DelayQueue。
  • 还有其他实现值得尝试吗？

​

• Storm是否有一些API来延迟我忽略的消息？
• ZeroMQ是否提供了Storm (如果修改)可以利用的延迟消息传递API？

Answer 1

使用外部消息队列实现时延队列.

由于Storm是容错的和水平分布的，因此选择适合这种样式的消息队列是有意义的，例如：

• 卡夫卡
• Amazon
• RabbitMQ

Answer 2

我们使用拓扑滴答元组来批量处理挂起的元组。它基本上只是将它们存储在每个普通元组上的内存中，当它接收到一个滴答元组时，它使用批量/流水线处理将它们处理到存储/索引中。

如果卷尖峰检测到每个主机上所有重定向到本地redis存储的元组，然后在卷停止后被推回拓扑处理，则我们也使用redis。我们的情况可能不适用于你的，只是我的2c。