Akka Actor - Warten Sie einige Zeit, um eine Nachricht zu erwarten, andernfalls senden Sie eine Nachricht aus

Question

Ist es möglich, eine Actor warten auf X Anzahl von Sekunden, um eine Nachricht zu empfangen, und wenn eine Nachricht empfangen wird, verarbeiten Sie es wie gewohnt, sonst senden Sie eine Nachricht an eine andere Actor (im Konstruktor festgelegt)?

Answer 1

Ja, wenn Sie für jede Nachricht warten möchten, können Sie einfach festlegen ein ReceiveTimeout: http://doc.akka.io/docs/akka/current/scala/actors.html#receive-timeout

(Die docs ist leicht irreführend hier, können Sie die receiveTimeout nach jeder Nachricht auch setzen)

Answer 2

Es ist möglich, werfen Sie einen Blick auf Akka Actor "ask" and "Await" with TimeoutException. Aber denken Sie daran, dass das Blockieren innerhalb eines Schauspielers eine sehr schlechte Idee ist, da der Schauspieler während dieser Zeit keine anderen Nachrichten verarbeiten kann. Außerdem blockiert es einen Akka-Verarbeitungs-Thread.

Ein besserer Ansatz ist es, eine Nachricht zu senden (Feuer und vergessen) und planen Sie ein Timeout-Ereignis mit Akka scheduler. Wenn die Antwort eintrifft, brechen Sie dieses Ereignis ab oder setzen Sie ein Flag, damit es nicht ausgelöst wird, wenn die Antwort tatsächlich pünktlich kam.

Answer 3

Könnte ein Overkill sein, aber Sie könnten die Finite State Machine (FSM) Eigenschaft überprüfen.

import akka._ 
import actor._ 
import util._ 
import duration._ 
import Impatient._ 

object Impatient { 
    sealed trait State 
    case object WaitingForMessage extends State 
    case object MessageReceived extends State 
    case object TimeoutExpired extends State 

    sealed trait Data 
    case object Unitialized extends Data 

    // In 
    case object Message 
} 

class Impatient(receiver: ActorRef) extends Actor with FSM[State, Data] { 
    startWith(WaitingForMessage, Unitialized) 

    when(WaitingForMessage, stateTimeout = 3 seconds) { 
    case Event(StateTimeout, data) => goto(TimeoutExpired) using data // data is usually modified here 
    case Event(Message, data) => goto(MessageReceived) using data // data is usually modified here 
    } 

    onTransition { 
    case WaitingForMessage -> MessageReceived => stateData match { 
     case data => log.info("Received message: " + data) 
    } 
    case WaitingForMessage -> TimeoutExpired => receiver ! TimeoutExpired 
    } 

    when(MessageReceived) { 
    case _ => stay 
    } 

    when(TimeoutExpired) { 
    case _ => stay 
    } 

    initialize 
} 

Hier ist es in Aktion:

object Main extends App { 
    import akka._ 
    import actor._ 
    import Impatient._ 

    val system = ActorSystem("System") 

    val receiver = system.actorOf(Props(new Actor with ActorLogging { 
    def receive = { 
     case TimeoutExpired => log.warning("Timeout expired") 
    } 
    })) 

    val impatient = system.actorOf(Props(new Impatient(receiver)), name = "Impatient") 
    impatient ! Message 

    val impatient2 = system.actorOf(Props(new Impatient(receiver)), name = "Impatient2") 
    Thread.sleep(4000) 
    impatient2 ! Message 

    system.shutdown() 
}