Как обрабатывать ответы от детей-актеров?

Question

Предположим, у нас есть актер, который делит задачи на подзадачи и делегирует их дочерним актерам. Этот актер выполняет свою собственную работу после получения ответов от всех субактеров.

Какова наилучшая практика для управления ответами субакторов?

Поместите свои результаты в, например, Map<ActorRef, Object> и проверьте каждый раз, если все ответы были получены?

Answer 1

Helo, как правило, я реализую его с помощью List, или если вы хотите Map, типа объектов. Либо: Left и Right (для сохранения информации об итогах каждого актера - успех или неудача) и проверка длины карты каждый раз, когда новое сообщение из принимается ребенок, и когда все полученные сообщения отправляются на самообслуживание WorkDone или sth с этим значением.

Где-то в сети я видел модель с отдельным актером, созданной для сбора всех сообщений образуют Чайлдс

superactor:

val n = 1000 
val collector = system.actorOf(Props(classOf[Collector],n) 
for(i<-1 to n){ 
    system.actorOf(Child.props()).!(DoSomeWork)(collector) 
} 

коллектор

class Collector(expectedMsgs: Int) extends Actor { 
    var results = someCollection 

    def endCollecting() = { 
     // You can add here some logic before returning data 
     context.parent ! results.toList //or whatever you want 
     context.become(afterWork) 
    } 

    def receive = { 
     case r: DataDone => 
      results += Right(r) 
      if(results.length>=expectedMsgs){ 
       endCollecting() 
      } 
     case r: DataFail => 
      results += Left(r) 
      if(results.length>=expectedMsgs){ 
       endCollecting() 
      } 
    } 

    def afterWork = { 
     case _ => 
      // for catch some late or mutiple messages 
    } 

} 

Также обратите внимание на http://doc.akka.io/docs/akka/snapshot/contrib/aggregator.html

Answer 2

Yo u можете использовать Composing Futures. Лучше использовать его в сочетании с Routing для детей. Что-то вроде:

import scala.concurrent._ 
import akka.pattern._ 
import akka.routing._ 
import akka.actor._ 
import akka.util._ 
import scala.concurrent.duration._ 


case class Req(i: Int) 
case class Response(i: Int) 

class Worker extends Actor { def receive = {case Req(i) => sender ! Response(i) }} 
class Parent extends Actor { 
    import context.dispatcher 
    implicit val timeout = Timeout(10, MINUTES) //timeout for response from worker 
    val router: ActorRef = 
     context.actorOf(RoundRobinPool(5).props(Props[Worker]), "router") 
    def processResults(rs: Seq[Response]) = {println(rs); rs} 
    def receive = { 
     case t: List[Req] => 
      Future.sequence(t.map(router ? _).map(_.mapTo[Response])).map(processResults) 
    } 
} 

Результаты:

scala> ActorSystem().actorOf(Props(classOf[Parent])) ! List(Req(1), Req(2), Req(3)) 

List(Response(1), Response(2), Response(3))