我有一个 Spring 应用程序,它使用一个小型 Akka actor 系统(使用 Java),其中我有一个MasterActor扩展 Akka 的系统AbstractActor,它初始化 aRouter并设置一些 worker actor。它还监视工人的生命周期。我想重启一个 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所有工作人员的消息,Master actor 尝试启动新的,这也会Terminated立即进入,进入无限循环。
检测这种情况的正确方法是什么?有没有办法让 Master actor 检测到 actor 系统正在关闭,这样 workers 就不会再次重启?
守着星空守着你
相关分类