Per evitare OOM, sto vincolando le dimensioni della cassetta postale di alcuni dei miei attori Akka 1.1.3 con un dispatcher condiviso condiviso. Per esempio:Gestisci cassetta postale con contorno attore Akka MessageQueueAppendFailedException
object Static {
val dispatcher = Dispatchers.newExecutorBasedEventDrivenWorkStealingDispatcher(
"customDispatcher",
1000,
BoundedMailbox(capacity = 10)
)
}
class MyActor extends Actor {
self.dispatcher = Static.dispatcher
...
}
vorrei reagire alla casella di posta traboccante così posso messaggio i produttori a monte per mettere in pausa (nota a margine: sembra tristemente simile actor.stop(), attendere, e actor.start() genererà un ActorStartException). Alcune perdite di dati sono accettabili tra la compilazione della coda e la coda che drena un po '.
capitolo di Akka su Dispatchers dice
Quando si cerca di inviare un messaggio al attore si getterà un MessageQueueAppendFailedException (“BlockingMessageTransferQueue trasferimento scaduta”) se il messaggio non può essere aggiunto alla casella di posta all'interno il tempo specificato da pushTimeout.
Dove posso rilevare questa eccezione?
La documentazione fa sembrare che ho bisogno di avvolgere ogni myActor ! message in un try/catch. È giusto? Mi piacerebbe davvero centralizzare la sua gestione. Il mio Supervisor può forse intercettarlo ed eseguire il mio gestore?
Qual è lo scopo di ispezionarlo? –
Non so se questo si applicherà al tuo caso specifico, ma se ti imbatti in situazioni in cui rischierai di sovraccaricare le caselle di posta, potresti prendere in considerazione un progetto "pull" in cui i tuoi attori di consumo chiedono lavoro ad un produttore piuttosto che al produttore Spammare i tuoi consumatori. –
@Viktor è per intraprendere azioni a monte, in particolare per mettere in pausa i consumatori (aggiungerò questo alla domanda). Dan ha ragione e considererò senz'altro il modello di tiro. Tuttavia mi piacerebbe farlo subito con poco rearchitecting e la perdita di dati è accettabile. – Bluu