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

Question

我有一个使用小型 Akka 演员系统（使用 Java）的 Spring 应用程序，其中我有一个 MasterActor，它扩展了 Akka 的 AbstractActor，它初始化了一个 Router 并设置了一些工人演员。它还监视工人的生命周期。如果某个 Worker 演员因某些 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 消息，并且主演员尝试启动新的演员，这也立即得到Terminated，进入无限循环。

检测这种情况的正确方法是什么？有没有办法让 Master Actor 检测到 Actor 系统正在关闭，从而不再重新启动工作人员？

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