scala - Akka - 公共(public)服务 Actor : Identify or Extension

Question

可以说我有一些其他 Actor 常用的服务层 Actor 。例如，存储和检索域对象的注册表服务:

case class DomainObject(id: UUID)

class Registry extends akka.actor.Actor {
  def receive: Receive = {
    case o: DomainObject => store(o) // save or update object

    case id: UUID => sender ! retrieve(id) // retrieve object and send it back
  }
}

我不想将此类注册表的实例显式传递给可能使用它的所有参与者。而不是它，我希望他们能够以某种方式“定位”它。

为此，我可以想到两种解决方案:

Identify消息:每个注册用户参与者都从一些配置中知道注册参与者名称，并能够向其发送识别消息。之后 AgentIdentity收到消息，我们可以走了:

val registryName = ... // some name
val registryId = ... // some id
var registry = _

def preStart() {
  context.actorSelection(registryName) ! Identify(registryId)
}

def receive: Receive = {
  case ActorIdentity(`registryId`, ref) => registry = ref
}

我不喜欢这种方式，因为在用户角色初始化之后，有一个阶段我们不知道系统等中是否有注册表，因此不知道我们是否能够操作。

Akka Extensions:我可以创建一个扩展，它将:

一种。在初始化时在给定的 Actor System 中创建 Registry Actor 的实例；

湾。通过 Extension 中的某种方法将此 Actor 返回给需要它的用户。

object RegistryKey extends ExtensionKey[RegistryExtension]

class RegistryExtesion(system: ExtendedActorSystem) extends RegistryKey {
  val registry = system.actorOf(Props[Registry], "registry")
}

问题是:哪种方法更好，Akka Extesions 是否可以用于此？

Answer 1

我认为扩展的想法是一个很好的想法，只要你的注册参与者总是在同一个 ActorSystem .

或者，使用 actorSelection (改编自 Remote Lookup ):

class RegistryClient extends Actor {
  val path = "/path/to/registry/actor"
  context.setReceiveTimeout(3.seconds)

  def sendIdentifyRequest(): Unit =
    context.actorSelection(path) ! Identify(path)

  def receive = {
    case ActorIdentity(`path`, Some(ref)) ⇒
      context.setReceiveTimeout(Duration.Undefined)
      context.become(active(ref))
    case ActorIdentity(`path`, None) ⇒
      throw new RuntimeException("Registry not found")
    case ReceiveTimeout ⇒ sendIdentifyRequest() 
  }

  def active(registry: ActorRef): Actor.Receive = {
    // use the registry
  }
}

这将适用于远程或本地 Actor 。

让我们看看扩展解决方案。 Actors are created asynchronously .因此，您的扩展构造函数在调用 actorOf 时不会失败如果actor初始化失败。

如果您想确定 Actor 未能初始化，那么一种方法是 ask Actor 将回应的东西和 Await一个回应。 Await将抛出 TimeoutException如果 Actor 没有回应。

class RegistryExtension(system: ExtendedActorSystem) extends Extension {
  val registry = system.actorOf(Props[Registry], "registry")
  implicit val timeout: Timeout = Timeout(500.millis)
  val f = registry ? "ping"    // Registry should case "ping" => "pong" 
  Await.result(f, 500.millis)  // Will throw a TimeoutException if registry fails 
                               // to respond
}

TimeoutException当你调用 RegistryExtension(system).registry 时会被抛出第一次。