Referenz eines Schauspielers bei Verwendung eines Routers abrufen

Question

Ich versuche, einen Ereignisstrom zu verarbeiten, der in Sitzungen "sessionisiert" werden kann. Der Plan ist, einen Pool von Akteuren zu verwenden, wobei ein einzelner Akteur aus dem Pool alle Ereignisse von einer Sitzung verarbeitet (der Grund ist, dass ich einen Sitzungszustand beibehalten muss). Es scheint mir, dass ich das ActorRef für einen bestimmten Schauspieler behalten müsste, der einer bestimmten Sitzung zugewiesen wurde. Allerdings, wenn ich ein Schauspieler bin mit Pool unter Verwendung:

val randomActor = _system.actorOf(Props[SessionProcessorActor].withRouter(RandomPool(100)), name = "RandomPoolActor") 

Dann wird in diesem Fall die randomActor bietet ActorRef auf den gesamten Pool, nicht auf die einzelnen Akteure im Pool. Wie könnte ich dann das erreichen, was ich oben erwähnt habe?

Eine Möglichkeit, die ich mir vorstellen kann, ist, die Referenz zurückzusenden, nachdem der Akteur aus dem Pool initialisiert wurde (würde wahrscheinlich etwas wie RandomPoolActor $ ab usw. aussehen). Diese Methode hat jedoch ein paar Probleme, von denen ich ein ask-Muster anstelle von tell verwenden muss, damit ich kein Ereignis aus derselben Sitzung verpasse.

Eine andere Möglichkeit, dies zu erreichen? Irgendein anderes Muster anzunehmen?

Answer 1

Sie könnten eine ConsistentHashingPool verwenden, die etwas ähnlich zu dem, was Sie suchen, tut. A ConsistentHashingRouter stellt sicher, dass jede Nachricht in demselben Akteur basierend auf einem hashKey endet. Dieser Schlüssel wäre Ihre Session-ID in Ihrem Szenario. Es ist nicht notwendig, ActorRefs oder andere Referenzen zu halten, um dies zu erreichen.

Es gibt mehrere Möglichkeiten, Ihre hashKey in Ihrem Code zu definieren. Ich würde empfehlen, eine Fallklasse zu erstellen, die ConsistentHashable erweitert. Danach müssen Sie die Methode consistentHashKey implementieren. Beispiel:

case class HashableEnvelope(yourMsgClass: YourMsgClass) extends ConsistentHashable { 
    override def consistentHashKey = yourMsgClass.sessionId 
} 

Dann können Sie Ihren Pool wie folgt definieren:

val pool = system.actorOf(Props[SessionProcessorActor].withRouter(ConsistentHashingPool(100))) 

Eine andere Sache ist zu erwähnen, dass der Router sicher, dass alle Nachrichten mit dem gleichen HashKey wird im selben Schauspieler am Ende, Es stellt jedoch nicht sicher, dass ein bestimmter Akteur nur Nachrichten für einen bestimmten HashKey empfängt. Es kann für mehrere HashKeys empfangen werden. Das sollte kein Problem sein, nur Ihre SessionProcessorActor sollte in der Lage sein, ein paar Hash Keys statt nur einer zu verarbeiten.

Der konsistente Hashalgorithmus entscheidet, welche Nachricht an jeden Akteur gesendet wird. Sie können auf Wikipedia lesen, wie es funktioniert: https://en.wikipedia.org/wiki/Consistent_hashing. So verteilen Sie Nachrichten in einer gleichmäßigen Weise sollten Sie die Anzahl der virtuellen Knoten in der Konfiguration erhöhen (Standard: 10):

akka.actor.deployment.default.virtual-nodes-factor = 1000 

Je nachdem, wie viele sessionIds und Schauspieler Sie haben, werden Sie diese Nachricht sehen, werden immer verteilt Gleichmäßiger.