Akka TypedActor - 如何正确处理带有上下文的异步响应答案

【问题标题】：Akka TypedActor - how to correctly handle async responses with contextAkka TypedActor - 如何正确处理带有上下文的异步响应
【发布时间】：2014-06-11 22:38:28
【问题描述】：

我已经开始在 Scala 中使用 TypedActors，并且在做一些非常简单的事情时遇到了问题：我希望 Actor A 调用 Actor B 上的一个方法并在 Actor A 上的匿名函数中处理结果，但要确保：

我的响应处理函数是线程安全的，例如不会与访问 Actor A 状态的任何其他线程同时运行
我的响应处理函数可以引用 Actor A 的上下文

我怎样才能（或可以）同时满足这两个要求？

例如，这个 actor 只想调用 otherActor 上返回 Future[Int] 的 API，并用结果更新它的状态，然后做一些需要它的 actor 上下文的事情：

class MyActorImpl extends MyActor {

  // my mutable state
  var myNumber = 0

  // method proxied by TypedActor ref:
  def doStuff(otherActor: OtherActor): Unit = {
    otherActor.doOtherStuff onSuccess {
      // oops this is no longer running in MyActorImpl..
      // this could be on a concurrent thread if we
      case i => processResult(i)
    }
  }

  private def processResult(i: Int): Unit = {
    myNumber = 0 // oops, now we are possibly making a concurrent modification
    println(s"Got $i")

    // fails with java.lang.IllegalStateException: Calling TypedActor.context
    // outside of a TypedActor implementation method!
    println(s"My context is ${TypedActor.context}")
  }
}

我在这里缺少什么？我是否需要编写我的处理程序来调用代理接口上定义的方法以保证单条目？如果我不想在接口上公开那个特定的“私有”方法（例如 processResult），那看起来会很丑。

这是一个可以在 Scala REPL 中运行的完整版本：

import akka.actor._
import scala.concurrent._

val system = ActorSystem("mySystem")
import system.dispatcher

trait OtherActor {
  def doOtherStuff(): Future[Int]
}


trait MyActor {
  def doStuff(otherActor: OtherActor): Unit
}

class OtherActorImpl extends OtherActor {
  var i = 0
  def doOtherStuff(): Future[Int] = {
    i += 1
    Future {i} 
  }
}

class MyActorImpl extends MyActor {

  // my mutable state
  var myNumber = 0

  // method proxied by TypedActor ref:
  def doStuff(otherActor: OtherActor): Unit = {
    otherActor.doOtherStuff onSuccess {
      // oops this is no longer running in MyActorImpl..
      // this could be on a concurrent thread if we
      case i => processResult(i)
    }
  }

  private def processResult(i: Int): Unit = {
    myNumber = 0 // oops, now we are possibly making a concurrent modification
    println(s"Got $i")

    // fails with java.lang.IllegalStateException: Calling TypedActor.context
    // outside of a TypedActor implementation method!
    println(s"My context is ${TypedActor.context}")
  }
}

val actor1: MyActor = TypedActor(system).typedActorOf(TypedProps[MyActorImpl])
val actor2: OtherActor = TypedActor(system).typedActorOf(TypedProps[OtherActorImpl])

actor1.doStuff(actor2)

【问题讨论】：

“我的 Future 将与我的 TypedActor 的调度程序一起完成，从而防止并发访问它的状态” - 我很确定这是错误的。调度程序可以（并且通常是）管理多个线程，因此可以进行并发访问。
嗯，好点子。在仔细阅读doc.akka.io/docs/akka/2.3.2/scala/futures.html#within-actors 之后，我发现它没有提到任何单项保证。我想我要修改这个问题......
@wingedsubmariner - 再次感谢。问题已被修改......并简化了。
FWIW，如果你遵循每个 Actor 负责改变自己的状态的规则，问题就会消失，所以 ActorA 会向 ActorB 发送消息请求改变状态。

标签： scala akka actor typedactor

【解决方案1】：

您将 Actor 的状态暴露给外部世界，这是一件非常糟糕的事情。看这里：http://doc.akka.io/docs/akka/2.3.3/general/jmm.html section Actors and shared mutable state 第 9-10 行描述了你的情况。

@philwalk 已经描述了如何解决这个问题：Akka TypedActor - how to correctly handle async responses with context

【讨论】：