Akka 等待actor的计算答案

【问题标题】：Akka wait for computation of actorAkka 等待actor的计算
【发布时间】：2015-05-02 11:57:02
【问题描述】：

我想创建一个 Actor 的最小示例，它向 Actor 发送消息，然后等待该 Actor 的响应。这个例子的原因是我想在我的论文中使用它来讨论其他语言特性（例如，期货）而不是纯粹的演员的用法。所以这里的重点是它必须是一个在处理其他任何事情之前等待消息的参与者。

我的想法是演示一个actor，它请求从磁盘读取文件，然后进行一些长时间的计算，然后等待读取完成。

到目前为止，我有以下内容：

import akka.actor.Actor
import akka.actor.ActorSystem
import akka.actor.Props
import scala.io.Source


case class FileContents(content: String)
class WorkerActor extends Actor
{
  def receive =
  {
    case "compute" =>
      println("Computing!")
      // Create actor to read the file
      val reader = context.actorOf(Props[ReadFileActor])
      reader ! ReadFile("/home/christophe/code/thesis-example/src/main/resources/file.txt")

      // Heavy computation
      Thread.sleep(5000)

    case FileContents(content) =>
      println("Got file content:\n" + content)
      // Continue computation.
  }
}

case class ReadFile(path: String)
class ReadFileActor extends Actor
{
  def receive =
  {
    case ReadFile(path) =>
      var contents: String = ""
      for (line <- Source.fromFile(path).getLines())
      {
        contents += line
      }
      sender ! FileContents(contents)
  }
}


object Main extends App
{
  val system = ActorSystem("HelloSystem")

  val worker = system.actorOf(Props[WorkerActor], name = "worker")

  worker ! "compute"
  worker ! "compute"
}

但是这里发生的是WorkerActor 接收到计算消息，然后启动子actor 来读取文件。在升沉计算之后，它改为接收第二个compute 消息。最后收到来自ReadFile actor 的两条消息。

我真正想要发生的是WorkerActor 收到compute 消息，进行繁重的计算，然后等到他收到FileContents 消息。只有在那之后它才能接收任何其他消息（即第二条compute 消息）。我已阅读文档并四处搜索示例，但似乎找不到任何内容。

免责声明：我没有使用 Akka，除了我论文中的这个小例子。

【问题讨论】：

标签： akka

【解决方案1】：

只需为多个 compute 消息创建多个 (pool of) 工作人员，而不是一个工作人员，例如：

object Main extends App {

  val system = ActorSystem("HelloSystem")

  val router = system.actorOf(RoundRobinPool(2).props(Props[Worker]), "router")

  router ! "compute"
  router ! "compute"

}

如果您希望在第一个工作人员之后启动第二个工作人员：

def receive = {
  case "compute" => ...
  case FileContents(content) =>
     println("Got file content:\n" + content)
     // Continue computation.
     context.parent ! "compute" //send to the pool 
}

...

//Main:
router ! "compute"

另一种选择是记住"compute" 中的sender（它将是您的Main）并将响应发送回顶层：

var main = _

def receive = {
  case "compute" => 
     ...
     main = sender
  case FileContents(content) =>
     ...
     main ! "ack" 
}

//Main:
(router ? "compute") foreach (_ => router ! compute)

如果你不喜欢这里的未来 - 你可以用 actor 重写它：

//Main
class MainActor extends Actor {
   def receive = {
      case "start" => router ! "compute"
      case "ack" => router ! "compute"
   }
}

附： Actor 内部的阻塞计算应该是 managed properly 否则它们可能会导致线程饥饿。

【讨论】：

【解决方案2】：

听起来你想要的是keep processing这条消息，直到你得到回复。换句话说block

class WorkerActor extends Actor {
  def receive = {
     case "compute" =>
     println("Computing!")
     // Create actor to read the file
     val reader = context.actorOf(Props[ReadFileActor])
     val future: Future[FileContents] = 
       (reader ? ReadFile("file.txt").mapTo[FileContents]

    // Heavy computation
    Thread.sleep(5000)
    // this will block current thread and thus the actor,
    // so it will not process any other messages until 
    // future is completed or time is out
    Await.result(future, timeout)   
  }
}

BUT，这在演员领域被视为a very bad thing。

【讨论】：