scala - 处理 future 的理解 Scala

Question

我有一个 Actor ，他收到一条消息并运行两个 future 。这些 future 可以并行运行，所以我想我可以使用 for comprehension 来运行两个 future 并将它们的结果组合成对发件人的一个响应。我可以自己从每个结果中获取结果，但我不知道在它们都完成后如何汇总它们。

def receive = {
    case "pcbStatus" => {
      val currentSender = sender
      //first future
      val wsf = (self ? "workhorseStats")(5 seconds)

      val psf = Future.traverse(context.children)(x => {
        (x ? "reportStatus")(5 seconds)
      });

      val combined = for {
        r1 <- wsf
        r2 <- psf
      } yield (r1, r2)



      combined.onComplete {
        case Success(result:Any) => {
          val response = new SomeCaseClass(r1,r2)
          println("YAY: " + response)
          currentSender ! response
        }
        case Failure(failure) => {
          println("FAIL: " + failure)
        }
      }
   }
}

Answer 1

我编写了一个小示例，说明我认为您正在尝试做的事情。首先是两个actor类:

class ParentActor extends Actor{
  import context._
  import akka.pattern.pipe
  implicit val timeout = Timeout(5 seconds)      

  override def preStart = {
    context.actorOf(Props[ChildActor], "child-a")
    context.actorOf(Props[ChildActor], "child-b")
  }    

  def receive = {
    case "foo" =>
      val fut1 = (self ? "bar").mapTo[Int]
      val fut2 = Future.traverse(context.children)(child => (child ? "baz").mapTo[Int])

      val aggFut = for{
        f1 <- fut1
        f2 <- fut2
      } yield SomeResult(f1, f2.toList)

      aggFut pipeTo sender

    case "bar" =>
      sender ! 2
  }
}

class ChildActor extends Actor{
  def receive = {
    case "baz" =>
      sender ! 1
  }
}

然后你可以用这段代码测试它:

implicit val timeout = Timeout(5 seconds)

val system = ActorSystem("foo")
val actor = system.actorOf(Props[ParentActor])
val result = actor ? "foo"

import system._
result onComplete{
  case tr => println(tr)
}

当您运行它时，它应该打印 Success(SomeResult(2,List(1, 1)))。

这里有几件事。首先，使用 mapTo 允许知道类型，而不必处理 Any。其次，pipeTo 是避免关闭发送方的一个很好的选择，它还稍微简化了代码。