如何检测akka actor终止是由于系统关闭并避免重新启动它

Question

我有一个 Spring 应用程序，它使用一个小型 Akka actor 系统（使用 Java），其中我有一个 MasterActor 扩展了 Akka 的 AbstractActor初始化一个 Router 并设置一些 worker actor。它还监视工人的生命周期。如果 Worker actor 由于某些 Exception 而恰好死了，我想重新启动它。

 public MasterActor(ActorPropsFactory actorPropsFactory) {
    this.actorPropsFactory = actorPropsFactory;

    int workers = Runtime.getRuntime().availableProcessors() - 1;

    List<Routee> routees = Stream.generate(this::createActorRefRoutee).limit(workers).collect(Collectors.toList());

    this.router = new Router(new ConsistentHashingRoutingLogic(getContext().system()), routees);
  }

  private ActorRefRoutee createActorRefRoutee() {
    ActorRef worker = getContext().actorOf(actorPropsFactory.create(getWorkerActorClass()));
    getContext().watch(worker);
    return new ActorRefRoutee(worker);
  }

  private void route(Object message, Supplier<String> routingKeySupplier) {
    String routingKey = routingKeySupplier.get();
    RouterEnvelope envelope = new ConsistentHashingRouter.ConsistentHashableEnvelope(message, routingKey);
    router.route(envelope, getSender());
  }

 @Override
  public Receive createReceive() {
    return receiveBuilder()
        .match(
            EventMessage.class,
            message -> this.route(message, () -> message.getEvent().getId().toString()))
        .match(
            Terminated.class,
            message -> {
              logger.info("WorkerActor {} terminated, restarting", message.getActor());
              // todo: detect whether the system is shutting down before restarting the actor
              router = router.removeRoutee(message.actor())
                             .addRoutee(createActorRefRoutee());
            })
        .build();
  }

我遇到的问题是，如果 Spring 应用程序无法启动。（例如，它无法连接到数据库，或者某些凭据不正确或其他），我从所有工作人员收到 Terminated 消息，Master actor 尝试启动新的，这也得到 Terminated 立即进入死循环。

检测这种情况的正确方法是什么？有没有办法让 Master actor 检测到 actor 系统正在关闭，这样 worker 就不会再次重启？

Answer 1

你不能为你的路由器设置一个监督策略，这样你就可以检查导致失败的异常类型吗？这样你也不需要手动重启你的工人。

编辑：

您这样设置 SupervisorStrategy：

private static SupervisorStrategy strategy=
    new OneForOneStrategy(
    10,
    Duration.ofMinutes(1),
    DeciderBuilder.match(ArithmeticException.class,e->SupervisorStrategy.resume())
    .match(NullPointerException.class,e->SupervisorStrategy.restart())
    .match(IllegalArgumentException.class,e->SupervisorStrategy.stop())
    .matchAny(o->SupervisorStrategy.escalate())
    .build());
final ActorRef router=
        system.actorOf(
        new RoundRobinPool(5).withSupervisorStrategy(strategy).props(Props.create(Echo.class)));

您可以在此处阅读更多相关信息：

Router Actor supervision

Fault tolerance in Akka

如何检测akka actor终止是由于系统关闭并避免重新启动它

How to detect akka actor termination is due to system shutdown and avoid restarting it

java

akka

akka-actor