Design corretto in akka. - Invio di messaggi

Question

Ho ricevuto alcuni post su come e perché akka non garantisce la consegna dei messaggi. Lo documentation, questo discussion e le altre discussioni sul gruppo lo spiegano bene.

Sono abbastanza nuovo per Akka e desidero conoscere il design appropriato per un caso. Ad esempio, dire che ho 3 attori diversi su macchine diverse. Uno è responsabile per i libri di cucina, l'altro per la storia e l'ultimo per i libri di tecnologia.

Ho un attore principale su un'altra macchina. Supponiamo che ci sia una query per l'attore principale per cercare se abbiamo qualche libro disponibile. L'attore principale invia richieste ai 3 attori remoti e si aspetta il risultato. Quindi faccio questo:

val scatter = system.actorOf(
     Props[SearchActor].withRouter(ScatterGatherFirstCompletedRouter(
       routees=someRoutees, within = 10 seconds)), "router") 
    implicit val timeout = Timeout(10 seconds) 
    val futureResult = scatter ? Text("Concurrency in Practice") 
    //  What should I do here?. 
    //val result = Await.result(futureResult, timeout.duration) line(a) 

In breve, ho inviato richieste a tutti e 3 gli attori remoti e aspetto il risultato in 10 secondi.

Quale dovrebbe essere l'azione?

1. Dire che non ottengo il risultato in 10 secondi, dovrei inviare una nuova richiesta a tutti di nuovo?
2. Cosa succede se il tempo di within precedente è prematuro. Ma non so pre-mano su quanto tempo potrebbe richiedere.
3. Cosa succede se il tempo within è stato sufficiente ma il messaggio è stato rilasciato.

Se non ottengo risposta nel tempo within e rispedisco di nuovo la richiesta. Qualcosa di simile, che rimanga asincrono:

futureResult onComplete{ 
    case Success(i) => println("Result "+i) 
    case Failure(e) => //send again 
} 

Ma sotto troppe domande, wont si tratti troppi thread sulla chiamata e ingombranti? Se si decommenta il valore line(a), diventa sincrono e il carico in esecuzione potrebbe non funzionare correttamente.

Dire che non ottengo risposta in 10 secondi. Se il tempo di within era prematuro, allora si verificava ancora un pesante calcolo inutile. Se il messsage è caduto, quindi 10 secondi di tempo prezioso sprecato. Nel caso, per esempio, sapevo che il messaggio è stato consegnato, probabilmente avrei aspettato una maggiore durata senza essere stato scettico allo .

In che modo le persone risolvono tali problemi? ACK? Ma poi devo memorizzare lo stato nell'attore di tutte le domande. Deve essere una cosa comune e sto cercando il giusto design.

Answer 1

Ho intenzione di provare a rispondere ad alcune di queste domande per voi. Non avrò risposte concrete per tutto, ma spero di poterti guidare nella giusta direzione.

Per i principianti, è necessario modificare la modalità di comunicazione della richiesta ai 3 attori che effettuano ricerche di libri. L'utilizzo di un ScatterGatherFirstCompletedRouter probabilmente non è l'approccio corretto qui. Questo router aspetterà solo una risposta da una delle rotte (la prima a rispondere), quindi la serie di risultati sarà incompleta in quanto non conterrà i risultati delle altre 2 rotte. C'è anche un BroadcastRouter, ma che non si adatta alle tue esigenze in quanto gestisce solo tell (!) e non ask (?). Per fare ciò che si vuole fare, un'opzione è inviare la richiesta a ciascun destinatario, ottenendo Futures per le risposte e quindi combinarle in un aggregato Future utilizzando Future.sequence.Un esempio semplificato potrebbe assomigliare a questo:

case class SearchBooks(title:String) 
case class Book(id:Long, title:String) 

class BookSearcher extends Actor{ 

    def receive = { 
    case req:SearchBooks => 
     val routees:List[ActorRef] = ...//Lookup routees here 
     implicit val timeout = Timeout(10 seconds) 
     implicit val ec = context.system.dispatcher 

     val futures = routees.map(routee => (routee ? req).mapTo[List[Book]]) 
     val fut = Future.sequence(futures) 

     val caller = sender //Important to not close over sender 
     fut onComplete{ 
     case Success(books) => caller ! books.flatten 

     case Failure(ex) => caller ! Status.Failure(ex) 
     } 
    } 
} 

Ora che non sta andando essere il nostro codice finale, ma è un'approssimazione di ciò che il campione è stato tentando di fare. In questo esempio, se una delle route downstream fallisce/va in timeout, colpiremo il nostro blocco Failure e anche il chiamante avrà un errore. Se tutti hanno successo, il chiamante otterrà invece l'elenco aggregato degli oggetti Book.

Ora sulle vostre domande. Innanzitutto, chiedi se dovresti inviare nuovamente una richiesta a tutti gli attori se non ricevi una risposta da uno dei percorsi entro il timeout. La risposta a questa domanda dipende da te. Permetteresti al tuo utente dall'altra parte di vedere un risultato parziale (cioè i risultati di 2 dei 3 attori), o deve sempre essere l'insieme completo di risultati ogni volta? Se la risposta è sì, si potrebbe ottimizzare il codice che sta inviando ai routees a guardare come questo:

val futures = routees.map(routee => (routee ? req).mapTo[List[Book]].recover{ 
    case ex => 
    //probably log something here 
    List() 
}) 

Con questo codice, se uno qualsiasi dei routees timesout o fallisce per qualsiasi motivo, un elenco vuoto di ' Libro sarà sostituito per la risposta invece del fallimento. Ora, se non puoi vivere con risultati parziali, puoi rispedire nuovamente l'intera richiesta, ma devi ricordare che probabilmente c'è qualcuno dall'altra parte che aspetta i risultati del libro e non vuole aspettare per sempre.

Per la seconda domanda, si chiede se il timeout è prematuro? Il valore di timeout selezionato sarà completamente a tua discrezione, ma molto probabilmente dovrebbe basarsi su due fattori. Il primo fattore verrà dal test dei tempi di chiamata delle ricerche. Scopri in media quanto tempo ci vuole e seleziona un valore basato su quello con un piccolo cuscino per sicurezza. Il secondo fattore è il tempo in cui qualcuno dall'altra parte è disposto ad aspettare i risultati. Potresti essere molto prudente nel tuo timeout, rendendolo come 60 secondi solo per essere sicuro, ma se c'è davvero qualcuno dall'altra parte in attesa di risultati, per quanto tempo sono disposti ad aspettare? Preferisco ricevere una risposta di errore che indica che dovrei riprovare invece di aspettare per sempre. Quindi, tenendo conto di questi due fattori, dovresti selezionare un valore che ti permetta di ottenere risposte in una percentuale molto alta del tempo senza tuttavia far attendere troppo a lungo il chiamante dall'altra parte.

Per la domanda 3, si chiede cosa succede se il messaggio viene interrotto. In questo caso, suppongo che il futuro per chiunque debba ricevere quel messaggio sarà scaduto solo perché non otterrà una risposta perché l'attore destinatario non riceverà mai un messaggio a cui rispondere. Akka non è JMS; non ha modalità di riconoscimento in cui un messaggio può essere inviato nuovamente un numero di volte se il destinatario non riceve e accetta il messaggio.

Inoltre, come puoi vedere dal mio esempio, concordo con il non bloccare il totale Future utilizzando Await. Preferisco usare i callback non bloccanti. Il blocco in una funzione di ricezione non è ideale in quanto l'istanza Actor interromperà l'elaborazione della sua cassetta postale fino al completamento dell'operazione di blocco. Utilizzando una richiamata non bloccante, si libera quell'istanza per tornare all'elaborazione della propria cassetta postale e consentire che la gestione del risultato sia solo un altro lavoro che viene eseguito nello ExecutionContext, disaccoppiato dall'attore che elabora la propria cassetta postale.

Ora, se si vuole davvero non sprecare comunicazioni quando la rete non è affidabile, è possibile esaminare l'Reliable Proxy disponibile in Akka 2.2. Se non si desidera seguire questa rotta, è possibile eseguirla manualmente inviando periodicamente messaggi di tipo ping ai percorsi. Se non si risponde in tempo, si contrassegna come inattivo e non si inviano messaggi finché non si ottiene un affidabile (in un tempo molto breve) ping da esso, un po 'come un FSM per ogni routee.Ognuno di questi può funzionare se è assolutamente necessario questo comportamento, ma è necessario ricordare che queste soluzioni aggiungono complessità e dovrebbero essere utilizzate solo se è assolutamente necessario questo comportamento. Se stai sviluppando software per banche e hai assolutamente bisogno di semantiche di consegna garantite, altrimenti le conseguenze finanziarie negative daranno risultati altrimenti con questo tipo di approccio. Sii solo giudizioso nel decidere se hai bisogno di qualcosa del genere perché scommetto il 90% delle volte che non lo fai. Nel tuo modello, l'unica persona probabilmente interessata dall'aspettare qualcosa che potresti aver già conosciuto non avrà successo è il chiamante dall'altra parte. Utilizzando callback non bloccanti nell'attore, non viene fermato dal fatto che qualcosa potrebbe richiedere molto tempo; è già passato al suo prossimo messaggio. È inoltre necessario fare attenzione se si decide di inviare nuovamente in caso di fallimento. Non si vuole inondare le cassette postali degli attori riceventi. Se si decide di inviare nuovamente, collegarlo a un numero fisso di volte.

Un altro possibile approccio se avete bisogno di questo tipo di semantica garantita potrebbe essere quello di esaminare Akka Clustering Model. Se si raggruppavano le rotte downstream e uno dei server non funzionava, tutto il traffico verrebbe indirizzato al nodo che era ancora attivo fino a quando l'altro nodo non si fosse ripristinato.