scala – akka：用于组合来自多个孩子的消息的模式

aggregator pattern旨在帮助解决这种情况.它提供了一个聚合器特征,您可以在actor中使用它来执行聚合逻辑.

想要执行聚合的客户端actor可以启动基于聚合器的actor实例,并向其发送一条消息,该消息将启动聚合过程.

应为每个聚合操作创建一个新的聚合器,并在发送回结果时终止(当它收到所有响应或超时时).

下面列出了这种模式的示例,该模式用于对由Child类表示的actor所持有的整数值求和. (请注意,所有孩子都不需要由同一个父actor监督：SummationAggregator只需要一个ActorRef集合.)

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使用此SummationAggregator,您可以执行以下操作：

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

然后在父接收的某个地方处理AggregationResult.