Message Queues - Hinten anstellen spart Wartezeit

Serverprozesse müssen ihre Aufgaben ohne Unterbrechung ausführen, anstatt aufeinander zu warten. Deshalb sprechen sie asynchron miteinander. Jeder spricht, wenn er etwas zu sagen hat, und hört zu, wenn er für neue Aufgaben bereit ist. Die Wartezeit verbringen die Nachrichten in sauber getrennten Warteschlangen.

Message-Queuing-Systeme

Beim Message Queuing ist die Arbeit auf drei Komponenten verteilt. Die erste Komponente, der Producer, ist für die Erstellung eines Jobs verantwortlich. Die Jobbeschreibung wird dann vom Producer in eine Nachricht verpackt und an den Message Store gesendet. Dieser nimmt die Nachricht entgegen und reiht sie in eine Warteschlange ein. Die dritte Komponente, der Consumer, holt sich dann die älteste Nachricht aus der Queue und führt den darin beschriebenen Job aus. Sobald eine Nachricht erfolgreich zugestellt wurde, wird sie aus der Queue gelöscht.

Message Queuing beschreibt also ein Verfahren, das die Erzeugung eines Jobs von seiner Ausführung trennt - eine Entkopplung der Systeme. Dabei wird nicht nur die Erzeugung von der Verarbeitung entkoppelt, sondern auch die zeitliche Abhängigkeit der Komponenten. Eine Nachricht kann durchaus längere Zeit in der Queue verweilen, bevor sie von einem Consumer abgeholt wird.

In der einfachsten Variante ist Message Queuing ein Einweg-Nachrichtensystem, d.h. die Nachrichten fließen immer vom Producer über die Queue zum Consumer. Das bedeutet, dass der Producer keine Rückmeldung vom Consumer erhält, ob ein Job bereits ausgeführt wurde oder überhaupt ausgeführt werden kann. Er erhält bestenfalls eine Rückmeldung, ob die Message in eine Queue eingereiht werden konnte. Benötigt er eine Rückmeldung vom Consumer, so muss eine zweite Queue als Antwortkanal eingerichtet werden. Auf dem Antwortkanal tauschen Producer und Consumer dann ihre Rollen.

Auch wenn eine bidirektionale Kommunikation über zwei Queues eingerichtet wurde, sind Producer und Consumer vollständig voneinander entkoppelt. Wird beispielsweise der Producer durch eine effizientere Anwendung ersetzt, so merkt der Consumer davon nichts - zumindest solange das Nachrichtenformat gleich bleibt. Denn der Consumer holt sich einfach die nächste Nachricht aus der Queue und verarbeitet sie. Wer die Nachricht dort abgelegt hat, ist für den Consumer irrelevant.

Ein wichtiger Vorteil des Message Queuing ist die zeitliche Entkopplung. Angenommen, der Producer erstellt die Jobs so schnell, dass der Consumer nicht mehr in der Lage ist, diese in angemessener Zeit abzuarbeiten. Dann können weitere Instanzen der Consumer-Anwendung gestartet werden, die dann gemeinsam die Jobs aus der Queue abarbeiten. Umgekehrt funktioniert das natürlich auch. Ist der Producer nicht schnell genug, um die Jobs zu erzeugen, können weitere Instanzen des Producers gestartet werden, um den Consumer zu entlasten.

Message Queue (MQ)

Das MQ-Modell, auch bekannt als Point-to-Point (P2P), ist ein One-to-One-Nachrichtensystem. Dies bedeutet jedoch nicht, dass eine Queue einen dedizierten Nachrichtenkanal zwischen zwei Anwendungen darstellt. In eine Queue können beliebig viele Producer ihre Nachrichten stellen und auf der anderen Seite können beliebig viele Consumer ihre Nachrichten abholen. Eine Message Queue garantiert lediglich, dass jede Nachricht nur genau einmal zugestellt wird. Welcher Producer die Nachricht eingestellt hat und welcher Consumer sie verarbeitet, ist nicht vorherbestimmt. Wird eine Nachricht an einen Consumer ausgeliefert, so erhalten alle anderen Consumer diese Nachricht nicht mehr.

Das MQ-Modell kann als Pull- oder Push-Verfahren implementiert werden. Beim Pull-Verfahren ist der Consumer selbst dafür verantwortlich, eine neue Nachricht abzuholen, sobald er freie Kapazität dafür hat. Beim Push-Verfahren registriert sich der Consumer bei der Message Queue und bekommt von dieser seine Nachrichten zugewiesen. Auf diese Weise kann die Message Queue auch als Load Balancer fungieren, um Aufgaben gleichmäßig auf mehrere parallele Arbeitsprozesse zu verteilen.

Ist zum Zeitpunkt des Nachrichteneingangs kein Consumer verfügbar, verbleibt die Nachricht so lange in der Queue, bis sie an einen Consumer ausgeliefert werden kann. Eine Message Queue kann immer dann eingesetzt werden, wenn eine Nachricht nur genau einmal verarbeitet werden darf, wie zum Beispiel bei einer Kontobuchung.

Publish Subscribe (Pub/Sub)

Im Publish-Subscribe-Kontext wird der Produzent als Publisher und der Konsument als Subscriber bezeichnet. Anstelle von Queues wird hier meist von Topics gesprochen. Ein Topic ist eine Queue, die durch ein Schlüsselwort, das Topic, identifiziert wird.

Will ein Publisher eine Nachricht veröffentlichen, so muss er dieser ein Topic hinzufügen. Denn nur anhand des Topics kann das Nachrichtensystem die Nachricht in der richtigen Queue bereitstellen. Ein Subscriber muss sich entsprechend beim Nachrichtensystem für ein Topic registrieren, um die dort veröffentlichten Nachrichten zu erhalten. Jede in einem Topic veröffentlichte Nachricht gilt erst dann als versendet und wird gelöscht, wenn alle registrierten Subscriber sie erhalten haben.

Wenn zum Zeitpunkt des Erhalts der Nachricht kein Subscriber registriert ist, gilt die Nachricht sofort als zugestellt und wird gelöscht. Das Verhalten kann jedoch je nach Implementierung unterschiedlich sein. So gibt es beispielsweise Nachrichtensysteme, die Nachrichten für einen registrierten Subscriber, der seine Verbindung verloren hat, zwischenspeichern. Sobald der Abonnent wieder online ist, werden die wartenden Nachrichten ausgeliefert.

Das Publish Subscribe Pattern ist ein One-to-Many- oder Many-to-Many-Nachrichtensystem, das sicherstellt, dass alle Nachrichten in der gleichen zeitlichen Reihenfolge zugestellt werden, in der sie empfangen wurden. Es wird in der Regel für Broadcasting verwendet, wenn eine Nachricht an mehrere Subscriber weitergeleitet werden soll. In diesem Fall verarbeitet jeder Teilnehmer die Nachricht für seine eigenen Zwecke. Wenn eine Nachricht nur einmal verarbeitet werden darf, ist Publish Subscribe die falsche Wahl.

Als Anwendungsbeispiel wird oft ein Aktienbroker genannt. Hier könnte man eine Reihe von Publishern anlegen, die die verschiedenen Aktienkurse überwachen und jede Änderung in einem Topic veröffentlichen. Eine Reihe von Subscribern kann sich dann für das Topic registrieren und zum Beispiel Kursprognosen berechnen, die Kurse in einer Datenbank persistieren oder grafisch aufbereiten.

Eine zentrale Komponente in diesem System ist der Exchange, der einzige Ansprechpartner für die Publisher. Ein Publisher sendet seine Nachrichten nicht direkt an eine Queue, sondern an den Exchange. Der Exchange verwaltet alle Queues und kennt deren Topics. Sobald er eine Nachricht erhält, sortiert er sie in die richtige Queue ein.

Vorteile von Message Queuing

Messages Queues sind u.a. dann sinnvoll, wenn eine Anwendung temporär mehr Nachrichten produziert, als die empfangende Anwendung verarbeiten kann. Auf diese Weise kann der Producer nicht durch einen überlasteten Consumer blockiert werden. Die Nachrichten werden einfach in der Queue zwischengespeichert, bis der Consumer frei ist.

Der Consumer kann durch den Einsatz einer Message Queue in geringem Maße entlastet werden, da er keine Push-Nachrichten mehr vom Producer erhält. Stattdessen kann der Consumer einfach die nächste Nachricht abholen, sobald er dazu bereit ist.

Entkopplung

Der Einsatz von Message Queues entkoppelt die Anwendungen voneinander. Producer- und Consumer-Anwendung müssen sich nicht mehr gegenseitig kennen, sondern interagieren nur noch mit der Message Queue. Durch die Entkopplung können beide Anwendungen in getrennten Projekten von unterschiedlichen Teams entwickelt werden.

Skalierbarkeit

Alle drei Komponenten (Producer, Queue, Consumer) können bei Bedarf unabhängig voneinander skaliert werden. So können beispielsweise, sobald sich Nachrichten in der Queue ansammeln, zusätzliche Consumer-Prozesse gestartet werden.

Wartbarkeit

Die Entwicklung mehrerer kleiner Anwendungen anstelle einer großen monolithischen Anwendung erhöht auch die Wartbarkeit. Die Erfahrung zeigt, dass kleinere Projekte leichter zu warten sind.

Asynchrone Kommunikation

Die Anwendungen kommunizieren asynchron. Dadurch wird verhindert, dass der Producer durch den Consumer blockiert wird. Durch das Caching kann der Consumer sogar offline sein, wenn die Nachricht gesendet wird.

Anwendungsgebiete

Entkopplung von Microservices

Die folgende Abbildung zeigt, wie im Publish Subscribe Pattern mehrere Microservices über einen Message Broker kommunizieren. Jeder Microservice konsumiert ein bestimmtes Topic, das hier farblich dargestellt ist. Ebenso sendet jeder Microservice unter seinem eigenen Topic Nachrichten an den Exchange, der diese in die Queues für das jeweilige Topic einreiht und so an die anderen Microservices weiterleitet

Temporär verfügbare Geräte

Wenn ein Microservice vorübergehend offline ist, werden seine Tasks in der Queue gesammelt und können zu einem späteren Zeitpunkt ausgeführt werden. Dies kann zum Beispiel der Fall sein, wenn der Service auf einem mobilen Gerät läuft, in einer abgelegenen Region mit schlechter Internetverbindung oder einfach während der Docker Container im Rechenzentrum neu gestartet wird.

Hier zeigt sich der Vorteil gegenüber einer herkömmlichen Web Service Umgebung. Würden sich die Services direkt gegenseitig aufrufen, würde jede Anfrage an einen gerade nicht verfügbaren Service mit einem Fehler abbrechen. Jeder einzelne Service müsste sich selbst um das Caching und den Neustart kümmern.

In einer AMQP-Umgebung hingegen kann jeder Service jeden anderen jederzeit ansprechen. Wenn ein Service nicht mit seiner Queue verbunden ist, warten die Nachrichten, bis er wieder online ist. Dann werden alle anstehenden Tasks abgearbeitet. Abgesehen von einer Verzögerung kann der Ausfall sogar unbemerkt bleiben.

Risiken und Nebenwirkungen

Vorsicht ist geboten, wenn ein Microservice nach einem Neustart unter keinen Umständen veraltete Nachrichten verarbeiten darf. Ein Beispiel ist eine Turbine in einem Windpark, die jede Minute ihren Status meldet. Das Wartungspersonal stellt nun fest, dass die Turbine seit einer Stunde konstant die gleiche Windgeschwindigkeit meldet. Das kann nicht sein, mit der Anlage stimmt etwas nicht.

Also senden sie einen Stoppbefehl über die Message Queue. Als dieser keine Wirkung zeigt, fährt ein Technikerteam hinaus, schaltet die Anlage manuell ab und fährt nach der Reparatur alle Systeme wieder hoch. Kaum läuft die Anlage wieder fehlerfrei, kommt der Stoppbefehl über die Queue und die Anlage schaltet ab.

Für kritische Steuerungen, die eine sofortige oder gar keine Reaktion erfordern, ist eine synchrone Kopplung, beispielsweise über REST, offensichtlich besser geeignet.

Lösungsansatz: Nachrichten mit Verfallsdatum

Eine Erweiterung des AMQP-Standards ermöglicht es, die verzögerte Verarbeitung von dringenden Nachrichten zu verhindern. In RabbitMQ heißt diese Lösung TTL (time to live) und kann auf eine Queue oder auf einzelne Nachrichten angewendet werden.

Wenn eine Nachricht in Queue A wartet, bis ihre TTL abgelaufen ist, dann gilt sie für Queue A als tot. Das bedeutet, dass sie nicht mehr zugestellt und bestätigt wird, sondern bei nächster Gelegenheit vom Server gelöscht wird. Wenn dieselbe Nachricht auch in einer anderen Queue B wartet, kann dort eine andere TTL gelten, so dass die Nachricht in Queue B am Leben bleibt.

Beispielsweise wird ein Stoppbefehl für Teile eines Maschinenparks vom Absender an den Exchange geschickt und von diesem parallel in die Queues zweier Maschinen A und B gestellt. Beide Maschinen sind jeweils 15 Sekunden beschäftigt, bevor sie auf den neuen Befehl reagieren. Queue A hat eine TTL von 20 Sekunden, Maschine A erhält den Befehl und stoppt. Queue B hat eine TTL von nur 10 Sekunden, also verpasst Maschine B den Befehl und das Personal erhält eine Fehlermeldung.

TTL ist jedoch nicht Teil des AMQP-Standards, sondern ein spezielles Feature der RabbitMQ-Implementierung. Im Folgenden wird RabbitMQ mit seinen zusätzlichen Funktionen näher vorgestellt.

RabbitMQ

RabbitMQ ist eine freie Implementierung des Advanced Message Queuing Protokolls. Die Software kann auf dem eigenen Server installiert werden, in den meisten Fällen reicht sogar das mitgelieferte Docker-Image aus.

Der aus dem Standard-Image gestartete Docker-Container gibt zwei Ports frei: Auf Port 5672 ist der AMPQ-Server erreichbar, auf 15672 läuft die Administrationsoberfläche. Hier können Exchanges, Bindings und Queues manuell eingerichtet werden.

Einige Client-Bibliotheken wie EasyNetQ übernehmen die Konfiguration automatisch. In diesem Fall kann man in der Administrationsoberfläche sehen, ob alles richtig eingestellt ist.

Queues

Besonders praktisch ist die Queue-Ansicht. Hier können die Nachrichten in einer Queue nicht nur eingesehen, sondern auch gelöscht und einzelne Nachrichten manuell eingefügt werden. Beim Testen einer Consumer-Anwendung wird man diesen Dialog am häufigsten verwenden.

Normalerweise erstellt eine Consumer-Anwendung ihre Queues selbst. Sofern nicht anders konfiguriert, existiert die Queue dann bis zum nächsten Neustart des Message Brokers. Andernfalls kann die Queue auch permanent sein, d.h. sie überlebt einen Neustart von RabbitMQ. Auch das Gegenteil ist möglich, dafür gibt es zwei Arten von temporären Queues. Auto-delete Queues verschwinden, nachdem sich der letzte Consumer ausgeloggt hat. Exclusive Queues sind mit genau einem Consumer verbunden und verschwinden, sobald dieser sich abmeldet.

Exchanges

Die Ansicht Exchanges listet alle Exchanges auf und ermöglicht das manuelle Anlegen neuer Exchanges. Je nach verwendeter Client-Bibliothek ist dies nur selten notwendig. EasyNetQ beispielsweise legt alle benötigten Exchanges und Queues automatisch an.

Bei der manuellen Konfiguration ist die Auswahl eines geeigneten Exchange-Typs wichtig.

• Ein Direct Exchange routet eingehende Nachrichten anhand ihres Routing-Keys in eine bestimmte Queue.
• Ein Topic Exchange verwendet ebenfalls den Routing Key, sendet die Nachrichten aber nicht nur an eine Queue, sondern an alle, deren Suchmuster mit dem Key übereinstimmt.
• Ein Fanout Exchange sendet eine Nachricht parallel an mehrere Queues, ohne sie zu filtern.
• Ein Header-Exchange ermöglicht komplexes Routing anhand mehrerer Attribute im Nachrichtenkopf.

Bestätigung

Normalerweise gilt eine Nachricht als angekommen, wenn sie vom Broker an den Empfänger ausgeliefert wurde. Wenn der Empfänger die Verarbeitung mit einem Fehler abbricht, ist die Nachricht bereits aus der Queue gelöscht.

RabbitMQ gibt dem Empfänger die Möglichkeit, die erfolgreiche Verarbeitung zu bestätigen. Erst dann wird die Nachricht aus der Queue entfernt. Im Fehlerfall lehnt der Empfänger die Nachricht ab, woraufhin sie wieder in die Queue gestellt wird. Alternativ kann ein Exchange als “Dead Letter Exchange” eingerichtet werden, der fehlerhafte Nachrichten zur Überprüfung weiterleitet.

Dead Letter Exchanges

Manchmal können Nachrichten nicht zugestellt werden. Häufige Ursachen sind ein abgelaufener Time-to-Live oder ein Verarbeitungsfehler auf der Empfängerseite. Manchmal hat die Queue auch eine Größenbeschränkung und läuft über. Für solche Fälle kann ein DLX (Dead Letter Exchange) deklariert werden, der die verlorenen Nachrichten auffängt.

Priorität

Mehrere Verbraucher, die von derselben Warteschlange versorgt werden, können unterschiedliche Prioritäten haben. Die weniger wichtigen Verbrauchende sollen nur dann bedient werden, wenn der wichtigste Verbrauchende ausfällt.

In RabbitMQ ist es möglich, jedem Konsumenten beim Start eine Priorität zuzuweisen. Eine Queue bedient dann nicht mehr alle ihre Consumer der Reihe nach, sondern nur noch die mit der höchsten Priorität. Erst wenn alle wichtigsten Consumer blockiert sind, werden die zweitwichtigsten reihum beliefert.

Apache Qpid

Das Open Source Projekt Qpid bietet ebenfalls APIs und Broker an, um AMQP in eigene Software zu integrieren. Der Einsatz von Qpid lohnt sich allerdings nur in sehr speziellen Anwendungsbereichen, in denen die Funktionen anderer Message Broker nicht ausreichen.

Im Gegensatz zu RabbitMQ stehen drei verschiedene Broker zur Auswahl. Für reine Java-Umgebungen ist der “Broker-J” interessant. Die native Implementierung heißt “Qpid C++ Broker”. Wenn es auf Skalierbarkeit ankommt, sollte der hochperformante “Dispatch Router” gewählt werden.

Mit Qpid Proton können nicht nur Clients entwickelt werden. Sie unterstützt bei Bedarf auch die Entwicklung eines eigenen Brokers. So flexibel Qpid letztlich ist, so steil ist die Lernkurve bis zum ersten lauffähigen System.

Apache ActiveMQ

ActiveMQ ist ein Java-basierter Broker. Entsprechende Client-Bibliotheken sind für verschiedene Plattformen verfügbar. Hervorzuheben ist die API “Apache.NMS.AMQP”, die den Anspruch erhebt, alle AMQP 1.0-kompatiblen Messaging-Systeme zu unterstützen. Ein Softwaresystem, das um den ActiveMQ-Broker herum entwickelt wurde, kann somit prinzipiell an jeden anderen Broker angedockt werden.

Weitere AMQP-Varianten

Solace PubSub+

PubSub+ des Herstellers Solace ist nicht Open Source. Limitiert auf 10.000 Nachrichten pro Sekunde kann die Software kostenlos genutzt werden. Support und ungebremste Performance verspricht die Enterprise Edition.

Azure Service Bus

Der Azure Service Bus ist ein Cloud-basierter Broker. Das bedeutet, dass eine Installation in einer Test-VM oder im eigenen Rechenzentrum nicht möglich ist. Er eignet sich daher nur für Anwendungen, die ohnehin auf die Azure-Plattform angewiesen sind.

Fazit

AMQP ist ein beliebtes Mittel zur Entkopplung von Microservices. Je nach Anwendungsfall stehen freie, proprietäre oder auf eine Cloud-Plattform spezialisierte Broker zur Verfügung.

Nicht umsonst ist RabbitMQ einer der beliebtesten Broker. Er ist einfach zu bedienen und kann sogar als Docker-Container auf jedem Developer-Notebook laufen. Dank einer wachsenden Community rund um RabbitMQ gibt es Bibliotheken für viele verschiedene Programmiersprachen. Bei proprietären Lösungen ist man hingegen auf die Schnittstelle des Herstellers angewiesen.

Weitere spannende Themen aus der adesso-Welt findet ihr in unseren bisher erschienen Blog-Beiträgen.