1

我有一个期货池，每个未来都使用相同的 akka 演员系统——系统中的一些演员应该是全球性的，有些只在一个未来使用。

  val longFutures = for (i <- 0 until 2 ) yield Future {
    val p:Page = PhantomExecutor(isDebug=true)
    Await.result( p.open("http://www.stackoverflow.com/") ,timeout = 10.seconds)
  }

PhantomExecutor 尝试使用一个共享的全局参与者（简单增量计数器）system.actorSelection

  def selectActor[T <: Actor  : ClassTag](system:ActorSystem,name:String) = {
    val timeout = Timeout(0.1 seconds)
    val myFutureStuff = system.actorSelection("akka://"+system.name+"/user/"+name)
    val aid:ActorIdentity = Await.result(myFutureStuff.ask(Identify(1))(timeout).mapTo[ActorIdentity],
      0.1 seconds)

    aid.ref match {
      case Some(cacher) =>
        cacher
      case None =>
        system.actorOf(Props[T],name)
    }
  }

但是在并发环境中，由于竞争条件，这种方法不起作用。

对于这个问题，我只知道一种解决方案——在拆分为期货之前创建全局参与者。但这意味着我无法封装来自顶级图书馆用户的大量隐藏工作。

4

2 回答 2

2

你是对的，确保首先初始化全局参与者是正确的方法。你不能将它们绑定到一个伴生对象并从那里引用它们，这样你就知道它们只会被初始化一次吗？如果你真的不能采用这种方法，那么你可以尝试这样的方法来查找或创建演员。它与您的代码类似，但它包含在遇到竞争条件时（最多只有最大次数）返回查找/创建逻辑（递归）的逻辑：

  def findOrCreateActor[T <: Actor : ClassTag](system:ActorSystem, name:String, maxAttempts:Int = 5):ActorRef = {
    import system.dispatcher
    val timeout = 0.1 seconds

    def doFindOrCreate(depth:Int = 0):ActorRef = {
      if (depth >= maxAttempts) 
        throw new RuntimeException(s"Can not create actor with name $name and reached max attempts of $maxAttempts") 

      val selection = system.actorSelection(s"/user/$name")
      val fut = selection.resolveOne(timeout).map(Some(_)).recover{
        case ex:ActorNotFound => None
      }
      val refOpt = Await.result(fut, timeout)

      refOpt match {
        case Some(ref) => ref
        case None => util.Try(system.actorOf(Props[T],name)).getOrElse(doFindOrCreate(depth + 1))
      }
    }

    doFindOrCreate()
  }

现在重试逻辑会在创建actor时触发任何异常，因此您可能希望进一步指定（可能通过另一个recover组合器）仅在它获得时递归InvalidActorNameException，但您明白了。

于 2014-12-02T18:11:51.347 回答
0

您可能需要考虑创建一个负责创建“反”角色的经理角色。这样，您将确保计数器参与者创建请求被序列化。

object CounterManagerActor {
  case class SelectActorRequest(name : String)
  case class SelectActorResponse(name : String, actorRef : ActorRef)
} 

class CounterManagerActor extends Actor {
  def receive = {
    case SelectActorRequest(name) => {
      sender() ! SelectActorResponse(name, selectActor(name))
    } 
  }

  private def selectActor(name : String) = {
    // a slightly modified version of the original selectActor() method
    ???
  }
}
于 2014-12-02T18:24:02.660 回答