获取或创建子 Akka actor 并确保 liveness

Question

我正在尝试使用 Akka 角色的层次结构来处理每个用户状态。有一个父actor拥有所有子actor，并以正确的方式处理get-or-create（参见a1、a2）：

class UserActorRegistry extends Actor {
  override def Receive = {
    case msg@ DoPerUserWork(userId, _) =>
      val perUserActor = getOrCreateUserActor(userId)
      // perUserActor is live now, but will it receive "msg"?
      perUserActor.forward(msg)
  }

  def getOrCreateUserActor(userId: UserId): ActorRef = {
    val childName = userId.toActorName
    context.child(childName) match {
      case Some(child) => child
      case None => context.actorOf(Props(classOf[UserActor], userId), childName)
  }
}

为了回收内存，UserActors 在一段空闲时间后过期（即计时器触发子actor调用context.stop(self)）。

我的问题是我认为“getOrCreateUserActor”和接收转发消息的子 Actor 之间存在竞争条件 - 如果子 Actor 在该窗口中过期，则转发消息将丢失。

有什么方法可以检测到这种边缘情况，或者重构 UserActorRegistry 以排除它？

Answer 1

我可以看到您当前的设计存在两个问题，这些问题使您面临您提到的竞争条件：

1) 将终止条件（计时器发送毒丸）直接发送给子actor。通过采用这种方法，孩子当然可以在一个单独的线程（在调度程序内）上终止，同时，一条消息已设置为在UserActorRegistry 演员中转发给它（在调度程序内的不同线程上） )。

2) 使用PoisonPill 终止子进程。 PoisonPill 用于优雅停止，允许首先处理邮箱中的其他消息。在您的情况下，您由于不活动而终止，这似乎表明邮箱中没有其他消息。我在这里看到PoisonPill 是错误的，因为在您的情况下，可能会在PosionPill 之后发送另一条消息，并且在处理PoisonPill 之后该消息肯定会丢失。

因此，我建议您将非活动子代的终止委托给UserActorRegistry，而不是在子代本身中执行。当您检测到不活动的情况时，向UserActorRegistry 的实例发送一条消息，指示需要终止特定的孩子。当您收到该消息时，通过stop 终止该子进程，而不是发送PoisonPill。通过使用以串行方式处理的UserActorRegistry 的单个邮箱，您可以帮助确保在您即将向其发送消息时不会并行终止孩子。

现在，您必须处理一个复杂的问题。停止演员是异步的。因此，如果您对孩子调用stop，则在处理DoPerUserWork 消息时它可能不会完全停止，因此可能会向其发送一条消息，该消息将丢失，因为它正在停止的过程中。您可以通过保留一些内部状态（列表）来解决此问题，该状态表示正在停止的子进程。当您停止一个孩子时，将其名称添加到该列表中，然后在其上设置DeathWatch（通过context watch child）。当您收到该孩子的Terminated 事件时，将其名称从被终止的孩子列表中删除。如果您收到一个孩子的名字在该列表中的工作，请将其重新排队以进行重新处理，最多可能达到最大次数，以免永远尝试重新处理。

这不是一个完美的解决方案；这只是识别您的方法中的一些问题，并推动解决其中一些问题的正确方向。如果您想查看此代码，请告诉我，我会一起制作一些东西。

编辑

回应您的第二条评论。我认为您无法查看ActorRef 的孩子并看到它当前正在关闭，因此需要该正在关闭的孩子列表。您可以增强 DoPerUserWork 消息以包含 numberOfAttempts:Int 字段并增加此字段，如果您看到目标子当前正在关闭，则将其发送回自身进行重新处理。然后，您可以使用 numberOfAttempts 来防止永远重新排队，在某个最大尝试次数处停止。如果您对依赖DeathWatch 感到不自在，您可以在正在关闭的儿童列表中的项目中添加一个生存时间组件。如果它们在列表中但在列表中的时间过长，您可以在遇到它们时修剪它们。