AKKA:用于组合来自多个子级的消息的模式

Question

这是我遇到的模式：

演员A有多个孩子C1、...、Cn。收到消息后，A将其发送给其每个子项，每个子项对消息进行一些计算，并在完成后将其发送回A。 然后，A想将所有孩子的结果结合起来传递给另一个演员。

这个问题的解决方案是什么样子的？或者这是一种反模式？在这种情况下，应该如何处理这个问题？

这是一个微不足道的例子，希望能说明我当前的解决方案。我担心的是重复的代码（达到对称性）;不能很好地扩展到"很多"儿童;并且很难看到发生了什么。

import akka.actor.{Props, Actor}
case class Tagged[T](value: T, id: Int)
class A extends Actor {
  import C1._
  import C2._
  val c1 = context.actorOf(Props[C1], "C1")
  val c2 = context.actorOf(Props[C2], "C2")
  var uid = 0
  var c1Results = Map[Int, Int]()
  var c2Results = Map[Int, Int]()
  def receive = {
    case n: Int => {
      c1 ! Tagged(n, uid)
      c2 ! Tagged(n, uid)
      uid += 1
    }
    case Tagged(C1Result(n), id) => c2Results get id match {
      case None => c1Results += (id -> n)
      case Some(m) => {
        c2Results -= id
        context.parent ! (n, m)
      }
    }
    case Tagged(C2Result(n), id) => c1Results get id match {
      case None => c2Results += (id -> n)
      case Some(m) => {
        c1Results -= id
        context.parent ! (m, n)
      }
    }
  }
}
class C1 extends Actor {
  import C1._
  def receive = {
    case Tagged(n: Int, id) => Tagged(C1Result(n), id)
  }
}
object C1 {
  case class C1Result(n: Int)
}
class C2 extends Actor {
  import C2._
  def receive = {
    case Tagged(n: Int, id) => Tagged(C2Result(n), id)
  }
}    
object C2 {
  case class C2Result(n: Int)
}

如果你认为代码看起来很糟糕，请放轻松，我刚刚开始学习 akka ;)

Answer 1

对于许多或不同数量的儿童演员，Zim-Zam建议的询问模式很快就会失控。

聚合器模式旨在帮助解决这种情况。它提供了一个聚合器特征，您可以在参与者中使用该特征来执行聚合逻辑。

想要执行聚合的

客户端参与者可以启动基于聚合器的执行组件实例，并向其发送一条消息，以启动聚合过程。

应为每个聚合操作创建一个新的聚合器，并在发回结果时终止（当它收到所有响应或超时时）。

下面列出了此模式的一个示例，用于对 Child 类表示的参与者持有的整数值求和。（请注意，它们不需要都是由同一父 actor 监督的子级：SumtionAggregator 只需要一个 ActorRefs 的集合。

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import akka.actor._
import akka.contrib.pattern.Aggregator
object Child {
  def props(value: Int): Props = Props(new Child(value))
  case object GetValue
  case class GetValueResult(value: Int)
}
class Child(value: Int) extends Actor {
  import Child._
  def receive = { case GetValue => sender ! GetValueResult(value) }
}
object SummationAggregator {
  def props = Props(new SummationAggregator)
  case object TimedOut
  case class StartAggregation(targets: Seq[ActorRef])
  case object BadCommand
  case class AggregationResult(sum: Int)
}
class SummationAggregator extends Actor with Aggregator {
  import Child._
  import SummationAggregator._
  expectOnce {
    case StartAggregation(targets) =>
      // Could do what this handler does in line but handing off to a 
      // separate class encapsulates the state a little more cleanly
      new Handler(targets, sender())
    case _ =>
      sender ! BadCommand
      context stop self
  }
  class Handler(targets: Seq[ActorRef], originalSender: ActorRef) {
    // Could just store a running total and keep track of the number of responses 
    // that we are awaiting...
    var valueResults = Set.empty[GetValueResult]
    context.system.scheduler.scheduleOnce(1.second, self, TimedOut)
    expect {
      case TimedOut =>
        // It might make sense to respond with what we have so far if some responses are still awaited...
        respondIfDone(respondAnyway = true)
    }
    if (targets.isEmpty)
      respondIfDone()
    else
      targets.foreach { t =>
        t ! GetValue
        expectOnce {
          case vr: GetValueResult =>
            valueResults += vr
            respondIfDone()
        }
      }
    def respondIfDone(respondAnyway: Boolean = false) = {
      if (respondAnyway || valueResults.size == targets.size) {
        originalSender ! AggregationResult(valueResults.foldLeft(0) { case (acc, GetValueResult(v)) => acc + v })
        context stop self
      }
    }
  }
}

要使用来自父Actor的求和聚合器，您可以执行以下操作：

context.actorOf(SummationAggregator.props) ! StartAggregation(children)

然后在父接收中的某个位置处理聚合结果。

Answer 2

您可以使用?而不是!子演员 - 这将导致子演员返回包含其（最终）结果的Future，即在您Await Future的结果之前，一切都仍然是非阻塞的。 然后，父参与者可以撰写这些Futures并将其发送给另一个参与者 - 它已经知道每个Future's身份，因此您无需担心标记每条消息，以便以后可以将它们重新按顺序排列。 这里有一个简单的例子，每个子项返回一个随机Double，并且您希望将第一个子项的返回值除以第二个子项的返回值（即顺序很重要）。

import scala.concurrent.duration._
import akka.actor.{Props, Actor}
import akka.pattern.{ask, pipe}
import akka.util.Timeout
class A extends Actor {
  val c1 = context.actorOf(Props[C], "C1")
  val c2 = context.actorOf(Props[C], "C2")
  // The ask operation involves creating an internal actor for handling 
  // this reply, which needs to have a timeout after which it is
  // destroyed in order not to leak resources; see more below.
  implicit val timeout = Timeout(5 seconds)
  def receive = {
    case _ => {
      val f1 = c1 ? "anything" // Future[Any]
      val f2 = c2 ? "anything" // Future[Any]
      val result: Future[Double] = for {
        d1 <- f1.mapTo[Double]
        d2 <- f2.mapTo[Double]
      } yield d1 / d2
  }
}
class C extends Actor {
  def receive = {
    case _ => // random Double
  }
}