scala - AKKA 中失败消息的重试/重放

Question

我在当前的 .NET 项目中使用 AKKA.NET。

我的问题是:经验丰富的 AKKA 开发人员如何使用最新的 Java 或 .NET AKKA 库实现失败时重播消息模式？

这里有一些更多细节。

我想确保失败的消息(即参与者收到的导致异常的消息)被重播/重试多次，每次之间有一定的时间间隔。通常情况下，actor 会通过失败的消息被丢弃来重新启动。

我编写了自己的小帮助方法来解决这个问题:

   public void WithRetries(int noSeconds, IUntypedActorContext context, IActorRef receiver, IActorRef sender, Object message, Action action)
    {
            try
            {
                action();
            }
            catch (Exception e)
            {
                context.System.Scheduler.ScheduleTellOnce(new TimeSpan(0, 0, noSeconds), receiver, message, sender);
                throw;
            }
        }
    }

现在我的 Actor 通常看起来像这样:

 Receive<SomeMessage>(msg =>
        {

            ActorHelper.Instance.WithRetries(-1, Context, Self, Sender, msg, () =>     {
              ...here comes the actual message processing  
            });
        });

我喜欢上面的解决方案，因为它很简单。但是，我不喜欢它在我的代码中添加了另一层间接层，并且如果我在很多地方使用此辅助方法，代码会变得更加困惑。此外它还有一些局限性。首先，重试次数不受辅助方法控制。它是由监管者的监管策略控制的，我认为这是困惑的。此外，时间间隔是固定的，而在某些情况下我希望每次重试都会增加时间间隔。

我更喜欢可以使用 HOCON 配置的东西。或者可以作为交叉关注点应用的东西。

我可以看到有关 AKKA for Scala、AKKA for Java 和 AKKA.NET 的各种建议。我看过路由器的示例、断路器的示例(例如 http://getakka.net/docs/CircuitBreaker#examples )等等。我也看到过一些使用与上面相同的想法的例子。但我有一种感觉，它应该更简单。也许它涉及 AKKA Persistence 和事件的一些使用。

所以重复我的问题:经验丰富的 AKKA 开发人员如何使用最新的 Java 或 .NET AKKA 库实现失败时重播消息模式？

Answer 1

我去年的某个时候研究过这个 - 我远离我的开发机器，所以无法检查，所以这一切都来自内存:

我似乎记得解决这个问题的方法是结合存储和监督策略以及生命周期 Hook :)

我认为你可以将你的子actor代码包装在try-catch中，然后在出现错误的情况下，存储消息并重新抛出异常，以便由主管和所有其他人处理通常的监督策略开始发挥作用。我认为您会继续而不是重新启动。然后在适当的生命周期消息(onresume？!)中取消存储消息，这应该意味着失败的消息将被再次处理。

现在这与您上面已经发布的内容并没有什么不同，所以希望有人有更好的解决方案:)