我用的是Apache Storm的BaseWindowBolt。但是，如果达到消息超时，我会看到重复的数据包。例如下面是我的测试配置。我正在使用Kafka spout从一个主题中读取数据。我有我的螺栓与翻滚窗口大小为2和消息超时秒为30。
现在我产生了1消息的主题，并等待超时发生，然后当我调试我的拓扑代码运行我的拓扑本地我看到的是相同的数据包是未来两次在元组窗口对象。
这是我想要的行为，还是我做错了什么？理想情况下，如果发生超时， Storm 应该只处理到达窗口的消息，即1.
我用的是Apache Storm的BaseWindowBolt。但是，如果达到消息超时，我会看到重复的数据包。例如下面是我的测试配置。我正在使用Kafka spout从一个主题中读取数据。我有我的螺栓与翻滚窗口大小为2和消息超时秒为30。
现在我产生了1消息的主题，并等待超时发生，然后当我调试我的拓扑代码运行我的拓扑本地我看到的是相同的数据包是未来两次在元组窗口对象。
这是我想要的行为，还是我做错了什么？理想情况下，如果发生超时， Storm 应该只处理到达窗口的消息，即1.
1条答案
按热度按时间m2xkgtsf1#
我知道你为什么会这样。这是因为消息超时到达并且最后一条消息仍然没有处理，所以 Storm 将认为该消息失败并且将重试，因此导致元组窗口中包括相同的数据包。为了解决这个问题，我们可以使用更长的消息超时秒数或更低的窗口计数。