我们有一个可以保存数据库的应用程序。如果保存失败，是否有方法重试失败的螺栓？我们可不想一路失败到最后。
l2osamch1#
您可以将输出“蝎子尾”流添加到螺栓。无论哪个螺栓开始重试过程，都将读取流。这将在拓扑中创建一个循环。其思想是，当发生故障时，您可以将一个信息包写入该流，并将元组传递到将开始重试的上游bolt。数据包包含重试所需的任何状态。
iqxoj9l92#
在Storm中没有内置的支持。但是，您可以编写自己的解决方案：不要确认（或失败）失败的元组，将其缓冲在内部数据结构中（即，成员变量；也许是一个 List )，然后从 execute() 继续处理中的更多元组 execute() 直到您想要重试（可能是某个计时器，例如，您可能想要获取当前时间戳或基于计数器重试）。在重试时，在处理新的输入元组之前，从缓冲区接收失败的元组并尝试插入到db中。如果再次失败，请再次插入缓冲区。如果插入成功，则确认缓冲元组并继续处理当前输入元组。你只需要考虑风暴的 MESSAGE_TIMEOUT . 重试所用的时间不能超过此值，因为如果元组未在超时值内得到确认，则storm会自动使源元组失败。
List
execute()
MESSAGE_TIMEOUT
2条答案
按热度按时间l2osamch1#
您可以将输出“蝎子尾”流添加到螺栓。无论哪个螺栓开始重试过程，都将读取流。这将在拓扑中创建一个循环。其思想是，当发生故障时，您可以将一个信息包写入该流，并将元组传递到将开始重试的上游bolt。数据包包含重试所需的任何状态。
iqxoj9l92#
在Storm中没有内置的支持。但是，您可以编写自己的解决方案：
不要确认（或失败）失败的元组，将其缓冲在内部数据结构中（即，成员变量；也许是一个
List)，然后从execute()继续处理中的更多元组execute()直到您想要重试（可能是某个计时器，例如，您可能想要获取当前时间戳或基于计数器重试）。在重试时，在处理新的输入元组之前，从缓冲区接收失败的元组并尝试插入到db中。如果再次失败，请再次插入缓冲区。如果插入成功，则确认缓冲元组并继续处理当前输入元组。
你只需要考虑风暴的
MESSAGE_TIMEOUT. 重试所用的时间不能超过此值，因为如果元组未在超时值内得到确认，则storm会自动使源元组失败。