Akka aktorzy, Futures i zamknięcia

Question

Przeczytałem w Akka docs, że jest niebezpieczne zamknięcie zmiennych z aktora zamykającego.

Warning

W tym przypadku trzeba starannie unikać zamykania przez referencyjne Zawierające aktora, to znaczy nie wywoływać metody na załączając aktora od wewnątrz anonimowej klasy aktorem. To spowoduje, że złamie hermetyzację aktora i może wprowadzić błędy synchronizacji oraz warunki wyścigu, ponieważ kod drugiego aktora zostanie zaplanowany jednocześnie na dla aktora zamykającego.

Teraz mam dwóch aktorów, z których jeden żąda czegoś od drugiego i robi coś z wynikiem. W tym przykładzie poniżej zestawiłem, aktor Accumulator pobiera numery od aktora NumberGenerator i dodaje je, raportując sumę po drodze.

Można to wykonać w co najmniej dwóch różnych sposobów, jak pokazano w tym przykładzie z dwoma różnymi otrzymać funkcji ( vs B). Różnica między nimi jest taka, że ​​ nie zamyka się nad zmienną licznika; zamiast tego oczekuje liczby całkowitej i podsumowuje ją, podczas gdy B tworzy Future, który zamyka ponad licznik i wykonuje sumę. Dzieje się tak w przypadku anonimowego aktora stworzonego tylko po to, aby móc go przetwarzać, jeśli dobrze rozumiem, jak to działa.

import com.esotericsoftware.minlog.Log 

import akka.actor.{Actor, Props} 
import akka.pattern.{ask, pipe} 
import akka.util.Timeout 
import akka.util.duration._ 

case object Start 
case object Request 


object ActorTest { 
    var wake = 0 

    val accRef = Main.actorSystem.actorOf(Props[Accumulator], name = "accumulator") 
    val genRef = Main.actorSystem.actorOf(Props[NumberGenerator], name = "generator") 

    Log.info("ActorTest", "Starting !") 

    accRef ! Start 
} 

class Accumulator extends Actor { 
    var counter = 0 

    implicit val timeout = Timeout(5 seconds) 

    // A: WITHOUT CLOSURE 
    def receive = { 
    case Start => ask(ActorTest.genRef, Request).mapTo[Int] pipeTo self 
    case x: Int => counter += x; Log.info("Accumulator", "counter = " + counter); self ! Start 
    } 
    // B: WITH CLOSURE 
    def receive = { 
    case Start => ask(ActorTest.genRef, Request).mapTo[Int] onSuccess { 
     case x: Int => counter += x; Log.info("Accumulator", "counter = " + counter); self ! Start 
    } 
    } 
} 

class NumberGenerator extends Actor { 
    val rand = new java.util.Random() 

    def receive = { 
    case Request => sender ! rand.nextInt(11)-5 
    } 
} 

Czy używanie w tym przypadku zamknięć jest absolutnie złe? Oczywiście mógłbym użyć AtomicInteger zamiast Int, lub w jakimś scenariuszu sieciowym używając, powiedzmy, netty, wystawić operację zapisu na kanale threadsafe, ale nie o to tutaj chodzi.

na ryzyko prosząc śmieszne: czy jest jakiś sposób na przyszłość za onSuccess wykonać w ten aktora zamiast anonimowego środkowej aktor bez definiowania sprawę w otrzymują funkcję?

EDIT

Mówiąc jaśniej, moje pytanie brzmi: Czy istnieje sposób zmusić serię Futures uruchomić w tym samym wątku jako danego aktora?

Answer 1

Problemem jest to, że onSuccess zamierza uruchomić w innym wątku niż nici aktora receive zamierza uruchomić w. Można użyć podejście pipeTo lub użyć Agent. Stworzenie counter an AtomicInteger rozwiąże problem, ale nie jest tak czyste - to znaczy, że łamie model Aktora.

Answer 2

Najprostszym sposobem realizacji takiego projektu jest za pomocą „ognia i zapomnij” semantyczny:

class Accumulator extends Actor { 
    private[this] var counter = 0 

    def receive = { 
    case Start => ActorTest.genRef ! Request 
    case x: Int => { 
     counter += x 
     Log.info("Accumulator", "counter = " + counter) 
     self ! Start 
    } 
    } 
} 

Takie rozwiązanie jest całkowicie asynchroniczny i nie potrzeba żadnego limitu czasu.