Ein zweiter Blick auf Nebenläufigkeit (Actor)
Beim letzten Mal haben wir darüber gesprochen, wie man Nachrichtenwarteschlangen nutzt, um Nachrichten zwischen Threads (Prozessen) zu übermitteln und so die Kommunikation zwischen Threads zu ermöglichen.
Gemeinsame Nachrichtenwarteschlange
Dies basiert auf der Annahme, dass Nebenläufigkeit lediglich den Durchsatz des Systems erhöht. Unter dieser Annahme erledigt jeder Worker die gleiche Arbeit, daher kann eine gemeinsame Nachrichtenwarteschlange verwendet werden – es spielt keine Rolle, wer die nächste Anweisung erhält.
Was ist, wenn wir nicht gleich sind?
Was ist, wenn wir ein System implementieren müssen, in dem die Worker asynchron arbeiten und jeweils spezifische Aufgaben erledigen (z. B. unser IP-Telefon)? Nehmen wir an, wir verwenden weiterhin die obige Methode, bei der alle Worker eine gemeinsame Nachrichtenwarteschlange teilen. Dann entsteht ein Problem: Nachrichten können nicht an bestimmte Worker gesendet werden.
Wie lässt sich dieses Problem lösen? Eine einfache, direkte Lösung besteht darin, jeder Nachricht eine Kennung hinzuzufügen, die den Empfänger der Nachricht angibt:
queue = Queue([('Send_to_worker_A', 'do_something1'),
('Send_to_worker_B', 'do_something2'),
('Send_to_worker_B', 'do_something3'),
('Send_to_worker_C', 'do_something4'),
('Send_to_worker_A', 'do_something5')])
Sieht so aus, als wäre das Problem gelöst, oder? Aber wir haben ein neues Problem eingeführt.
Was, wenn ich versehentlich den Brief eines anderen öffne?
Wenn eine Nachricht für Worker A von Worker B empfangen wird, ist sie für diesen nutzlos. Der eigentlich vorgesehene Empfänger, Worker A, erhält sie nicht, und die Nachricht geht verloren. Das verschwendet nicht nur die Zeit von Worker B, sondern auch die von Worker A und könnte das gesamte System aufgrund eines fehlenden Schritts in seltsame Bugs stürzen.
Wie lässt sich dieses Problem lösen? Intuitiv betrachtet scheint es auszureichen, jedem Worker die folgende Strategie hinzuzufügen: Wenn er eine Nachricht für jemand anderen erhält, legt er sie zurück in die Warteschlange.
class Worker_B:
def run(self):
while True:
if not self.queue.empty():
data = self.queue.get()
if data['to_whom'] == 'worker_B':
do_something_with_data
else:
self.queue.put(data)
Es scheint, als hätten wir das Problem mit Flickwerk gelöst… aber tatsächlich haben wir ein neues Problem eingeführt.
Das sieht aus wie eine Sackgasse
Erstens ist eine Nachrichtenwarteschlange eine Warteschlange, was FIFO (First In, First Out) bedeutet. Selbst wenn wir eine priorisierte Warteschlange einführen, ändert sich dieser grundlegende Punkt nicht. Dadurch wird die Nachricht, die eigentlich als erste an Worker A gehen sollte, nachdem Worker B sie “umgeleitet” hat, zur letzten in der Warteschlange für Worker A. Dies könnte das System in seltsame Bugs stürzen.
Zweitens: Selbst wenn wir es schaffen, die falsch verteilte Nachricht wieder an den Anfang der Warteschlange zu setzen, müssen wir, sofern wir keinen Sperrmechanismus einführen, der den Vorgang Nachricht_empfangen–Prüfen–Zurücklegen von Worker B zu einem atomaren Vorgang macht, damit rechnen, dass andere Worker bereits mit dem Empfang von Nachrichten fortfahren, während Worker B die Nachricht zurücklegt. Dies könnte weiterhin zu erheblichen zeitlichen Problemen führen. Die Einführung eines Sperrmechanismus könnte zwar logische Fehler beheben, hätte aber zur Folge, dass bei jedem Abfragezyklus nur ein Worker tatsächlich arbeiten könnte, während andere, die eine falsche Nachricht erhalten und den Vorgang Empfangen–Prüfen–Zurücklegen ausführen, ihre Zeitscheibe verschwenden – ineffizient.
Daher scheint es für Systeme, die Nebenläufigkeit für asynchrone Arbeit einführen, keine gute Idee zu sein, dass alle Worker dieselbe Nachrichtenwarteschlange teilen.
Die radikale Lösung für dieses Problem besteht darin, für jede Gruppe von Workern, die die gleiche Aufgabe ausführen, eine eigene, spezielle Nachrichtenwarteschlange (Postfach) einzurichten, oder sogar für jeden Worker ein eigenes Postfach, wodurch die Fähigkeit der Threads, Speicher gemeinsam zu nutzen, vollständig aufgegeben wird. Die letztere Methode ist der heutige Hauptdarsteller: das Actor-Modell (Akteur-Modell).
Das Actor-Modell
Schauen wir uns zunächst die Definition und Konzepte des Actor-Modells laut Wikipedia an:
Konzept
In der Informatik ist das Akteurmodell (englisch: Actor model) ein Modell für nebenläufige Berechnungen. Ein “Akteur” ist ein abstraktes Programmierkonzept, das als grundlegende Einheit der nebenläufigen Berechnung betrachtet wird: Wenn ein Akteur eine Nachricht empfängt, kann er Entscheidungen treffen, weitere Akteure erstellen, weitere Nachrichten senden und festlegen, wie er auf die nächste Nachricht reagieren wird. Das Akteurmodell wurde 1973 in einem Artikel von Carl Hewitt, Peter Bishop und Richard Steiger vorgestellt.
Die Philosophie des Akteurmodells ist “Alles ist ein Akteur”, ähnlich wie “Alles ist ein Objekt” in der objektorientierten Programmierung, jedoch ist die objektorientierte Programmierung in der Regel sequentiell, während das Akteurmodell parallel ausgeführt wird. Ein Akteur ist eine Berechnungseinheit, die auf empfangene Nachrichten reagiert und dabei parallel:
- eine begrenzte Anzahl von Nachrichten an andere Akteure sendet;
- eine begrenzte Anzahl neuer Akteure erstellt;
- das Verhalten für den Empfang der nächsten Nachricht festlegt.
Bei diesen Operationen wird keine sequentielle Ausführung angenommen, daher können sie parallel durchgeführt werden. Die Entkopplung des Senders von der gesendeten Nachricht ist der grundlegende Vorteil des Akteurmodells. Dies ermöglicht asynchrone Kommunikation und erfüllt gleichzeitig die Kontrollstrukturen der Nachrichtenübermittlung. Nachrichtenempfänger werden über Adressen unterschieden, die manchmal auch als “E-Mail-Adressen” bezeichnet werden. Daher kann ein Akteur nur mit Akteuren kommunizieren, deren Adresse er besitzt. Er kann Adressen über empfangene Informationen erhalten oder die Adressen der von ihm erstellten Akteure abrufen. Charakteristisch für das Akteurmodell ist, dass Berechnungen innerhalb von Akteuren oder zwischen ihnen parallel ablaufen, Akteure dynamisch erstellt werden können, Adressen in Nachrichten enthalten sind, die Interaktion nur über direkte asynchrone Nachrichtenkommunikation erfolgt und die Reihenfolge des Nachrichteneingangs nicht eingeschränkt ist.
Implementierung
Akteure sind einfach genug, für Nebenläufigkeit geschaffen und bieten eine ausreichend gute Kapselung, um Änderungen zu isolieren (z. B. unabhängig davon, ob es sich um Multithreading oder Multiprocessing handelt).
Schauen wir uns eine typische Python-Thread-Implementierung eines Akteurs an:
from queue import Queue
from threading import Thread, Event
class WorkerExit(Exception):
# Ausnahme zum Beenden der Aufgabe
pass
class BaseWorker(object):
# Actor
def __init__(self):
self._mailbox = Queue()
super(BaseWorker, self).__init__()
def send(self, msg):
self._mailbox.put(msg)
def recv(self):
msg = self._mailbox.get()
if msg is WorkerExit:
raise WorkerExit()
return msg
def close(self):
self.send(WorkerExit)
def start(self):
self._terminated = Event()
t = Thread(target=self._bootstrap)
t.daemon = True
t.start()
def _bootstrap(self):
try:
self.run()
except WorkerExit:
pass
finally:
self._terminated.set()
def join(self):
self._terminated.wait()
def run(self):
raise NotImplementedError
Kurz erklärt: Dieser Akteur (ich habe ihn BaseWorker genannt, in echten Projekten wird er als Elternklasse verwendet) verwaltet zwei Datenstrukturen: eine Queue als Postfach und ein Event als Mechanismus zum Blockieren des Hauptthreads.
Die zentrale Operation, die hier als externe Schnittstelle dient, ist nur die Methode send(). Beachten Sie, dass wir den Typ der übermittelten Nachricht nicht einschränken, was eine enorme Flexibilität bedeutet.
Innerhalb des Akteurs verwenden wir einen Thread, um die Methode run() in Kombination mit der Methode recv() auszuführen und die spezifische Arbeit zu erledigen. Erwähnenswert ist, dass wir ein Sentinel-Signal WorkerExit zum Stoppen der Aufgabe eingeführt haben. Beachten Sie, dass WorkerExit funktioniert, indem es, sobald es erkannt wird, als Ausnahme ausgelöst wird. In der Ausnahmebehandlung könnten wir hier sogar mehr tun, aber in diesem Fall beenden wir einfach den Thread, wenn diese Ausnahme abgefangen wird. Die Implementierung dieser Ausnahmebehandlung hängt von der Methode _bootstrap() ab, die run() umschließt.
Indem wir dieses einfache Beispiel im Sinne der Philosophie des Akteurmodells erweitern, können wir noch weiter gehen.
Zusammenfassung
Nebenläufige Programmierung erfordert eine abstraktere Denkweise als gewöhnlich und kann auch als gute Praxis für das Prinzip “hohe Kohäsion, lose Kopplung” dienen.
In meiner Erkundung dieses für mich völlig neuen Gebiets war das Actor-Modell tatsächlich ein Leitstern, der mich wirklich eingeführt hat, daher teile ich es hier. :)