获取或创建儿童Akka演员并确保活力

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之间存在竞争条件 - 如果子进程在该窗口中到期,则转发的消息将丢失.

有什么方法可以检测到这种边缘情况,还是重构UserActorRegistry以排除它？

Answer 1

我可以看到你当前设计的两个问题,让你自己达到你提到的竞争条件:

1)终止条件(计时器发送毒丸)直接进入儿童演员.通过采用这种方法,子节点当然可以在一个单独的线程上(在调度程序内)终止,同时,消息已经被设置为在UserActorRegistryactor 中转发给它(在调度程序内的不同线程上).

2)使用a PoisonPill来终止孩子.A PoisonPill用于正常停止,允许首先处理邮箱中的其他邮件.在您的情况下,由于不活动而终止,这似乎表明邮箱中已有其他邮件.我PoisonPill在这里看到一个错误,因为在你的情况下,可能会发送另一条消息,PosionPill并且该消息肯定会在PoisonPill处理后丢失.

因此,我建议您将非活动儿童的终止委托给他们UserActorRegistry,而不是在儿童自己中进行.当您检测到不活动状态时,向实例发送消息,UserActorRegistry指示需要终止特定子项.当您收到该消息时,请通过stop而不是发送一个来终止该子节点PoisonPill.通过使用UserActorRegistry以串行方式处理的单个邮箱,您可以帮助确保在您要向其发送消息时不会并行终止子项.

现在,这里有一个复杂的问题,你必须处理.停止一个actor是异步的.因此,如果您stop对孩子进行呼叫,则在处理DoPerUserWork消息时可能不会完全停止,因此可能会向其发送一条消息,因为它正在停止.您可以通过保留一些表示正在停止的子项的内部状态(List)来解决此问题.当您停止孩子时,将其名称添加到该列表,然后在其上设置DeathWatch(通过context watch child).当您收到该Terminated孩子的活动时,请从被终止的孩子列表中删除该名称.如果您的名字在该列表中时为您的孩子收到工作,请将其重新排队以进行重新处理,最多可能达到最大次数,以免永远尝试重新处理.

这不是一个完美的解决方案; 它只是对您的方法的一些问题的识别,并推动正确的方向来解决其中的一些问题.如果你想看到这个的代码,请告诉我,我会把东西放在一起.

编辑

回应你的第二条评论.我不认为你能看到一个孩子ActorRef,看到它正在关闭,因此需要关闭正在关闭的孩子名单.您可以增强DoPerUserWork消息以包含numberOfAttempts:Int字段并将其递增并发送回self以进行重新处理(如果您看到目标子节点当前正在关闭).然后,您可以使用numberOfAttempts来防止永久重新排队,在某些最大尝试次数停止.如果您感觉不完全依赖DeathWatch,可以在关闭的子项列表中为项目添加生存时间组件.然后你可以在遇到它们时修剪项目,如果它们在列表中但在那里已经存在太久了.