🥇Bausteine verteilter Anwendungen. Erste Annäherung

Im letzten Artikel Wir haben die theoretischen Grundlagen der reaktiven Architektur erörtert. Jetzt ist es an der Zeit, über Datenströme, Implementierungswege von reaktiven Erlang/Elixir-Systemen und die Muster des Nachrichtenaustauschs zu sprechen:

Anfrage-Antwort
Anfrage-Chunked Antwort
Antwort mit Anfrage
Veröffentlichen-Abonnieren
Umgekehrtes Veröffentlichen-Abonnieren
Taskverteilung

SOA, MSA und Nachrichtenaustausch

SOA, MSA – Systemarchitekturen, die die Regeln für den Aufbau von Systemen definieren, während Messaging primitives für deren Implementierung bereitstellt.

Ich möchte keine bestimmte Architektur propagieren. Ich bin für die Anwendung von möglichst effektiven und nützlichen Praktiken für das jeweilige Projekt und Unternehmen. Welche Paradigmen wir auch wählen, Systemkomponenten sollten im Sinne des Unix-Ways erstellt werden: Komponenten mit minimaler Kopplung, die für einzelne Entitäten verantwortlich sind. API-Methoden führen die einfachsten Aktionen mit Entitäten aus.

Messaging ‒ wie der Name schon sagt ‒ ist ein Nachrichtenbroker. Sein Hauptziel ist es, Nachrichten entgegenzunehmen und zu übermitteln. Er sorgt für Schnittstellen zur Informationsübertragung, die Bildung logischer Übertragungskanäle innerhalb des Systems, Routing und Lastverteilung sowie die systemweite Fehlermanagement.
Das entwickelte Messaging-System soll nicht mit RabbitMQ konkurrieren oder es ersetzen. Seine Hauptmerkmale sind:

Dezentralisierung.
Austauschpunkte können an allen Knoten des Clusters erstellt werden, so nah wie möglich am Code, der sie verwendet.
Benutzerfreundlichkeit.
Fokus auf Minimierung von Boilerplate-Code und Benutzerfreundlichkeit.
Hervorragende Leistung.
Wir versuchen nicht, die Funktionalität von RabbitMQ zu replizieren, sondern heben nur die Architektur- und Transportschicht hervor, die wir möglichst einfach in OTP integrieren, um die Kosten zu minimieren.
Flexibilität.
Jeder Service kann zahlreiche Austauschmuster kombinieren.
Fehlertoleranz, die in das Design integriert ist.
Skalierbarkeit.
Messaging wächst mit der Anwendung. Bei zunehmender Last können Austauschpunkte auf separate Maschinen verlagert werden.

Hinweis. Aus Sicht der Code-Organisation sind Meta-Projekte für komplexe Systeme in Erlang/Elixir gut geeignet. Der gesamte Projektcode befindet sich in einem Repository – einem übergeordneten Projekt. Dabei sind Mikroservices maximal isoliert und führen einfache Operationen aus, die eine separate Entität betreffen. Mit diesem Ansatz lässt sich die API des gesamten Systems einfach warten, Änderungen sind leicht umzusetzen, und es ist bequem, Unit- und Integrationstests zu schreiben.

Die Komponenten des Systems interagieren direkt oder über einen Broker. Aus der Perspektive des Messaging hat jeder Service mehrere Lebensphasen:

Initialisierung des Services.
In dieser Phase erfolgt die Konfiguration und der Start des ausführenden Service-Prozesses und der Abhängigkeiten.
Erstellung eines Austauschpunkts.
Der Service kann einen statischen Austauschpunkt verwenden, der in der Knoten-Config festgelegt ist, oder Austauschpunkte dynamisch erstellen.
Registrierung des Services.
Damit der Service Anfragen bedienen kann, muss er am Austauschpunkt registriert werden.
Normale Funktionalität.
Der Service führt nützliche Arbeit aus.
Beendigung der Arbeit.
Es sind zwei Arten des Herunterfahrens möglich: regulär und notfallmäßig. Bei einer regulären Abschaltung wird der Dienst vom Austauschpunkt getrennt und gestoppt. In Notfällen führt das Messaging eines der Ausfallszenarien aus.

Es sieht ziemlich kompliziert aus, aber im Code ist nicht alles so schlimm. Beispielcodes mit Kommentaren werden im Template-Review etwas später bereitgestellt.

Austausche

Der Austauschpunkt ist ein Messaging-Prozess, der die Logik der Interaktion mit Komponenten im Rahmen des Nachrichtenaustausch-Templates implementiert. In allen unten präsentierten Beispielen interagieren die Komponenten über Austauschpunkte, deren Kombination das Messaging bildet.

Message Exchange Patterns (MEPs)

Global betrachtet lassen sich Austausch-Templates in bidirektionale und unidirektionale unterteilen. Erstere unterstellen eine Antwort auf die empfangene Nachricht, letztere nicht. Ein klassisches Beispiel für ein bidirektionales Template in der Client-Server-Architektur ist das Anfrage-Antwort-Template. Lassen Sie uns das Template und seine Modifikationen betrachten.

Anfrage-Antwort oder RPC

RPC wird verwendet, wenn wir eine Antwort von einem anderen Prozess benötigen. Dieser Prozess kann auf demselben Knoten oder auf einem anderen Kontinent ausgeführt werden. Im Folgenden finden Sie das Schema der Interaktion zwischen dem Client und Server über Messaging.

Da Messaging vollständig asynchron ist, wird der Austausch in zwei Phasen unterteilt für den Client:

Anfrage senden
```
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
Exchange ‒ ein einzigartiger Name für den Austausch
ResponseMatchingTag ‒ ein lokales Tag zur Verarbeitung der Antwort. Zum Beispiel, wenn mehrere identische Anfragen gesendet werden, die verschiedenen Benutzern gehören.
RequestDefinition ‒ der Inhalt der Anfrage
HandlerProcess ‒ PID des Handlers. Dieser Prozess erhält die Antwort vom Server.

Antwortverarbeitung

handle_info(#'$msg'{exchange = EXCHANGE, tag = ResponseMatchingTag, message = ResponsePayload}, State)

ResponsePayload ‒ die Serverantwort.

Für den Server besteht der Prozess auch aus zwei Phasen:

Initialisierung des Austauschnodes
Verarbeitung eingehender Anfragen

Veranschaulichen wir dieses Muster mit Code. Angenommen, wir müssen einen einfachen Dienst implementieren, der eine einzige Methode zur genauen Zeit bereitstellt.

Servercode

Lassen Sie uns die Definition der API des Dienstes in api.hrl auslagern:

%% =====================================================
%%  Entities
%% =====================================================
-record(time, {
  unixtime :: non_neg_integer(),
  datetime :: binary()
}).

-record(time_error, {
  code :: non_neg_integer(),
  error :: term()
}).

%% =====================================================
%%  Methods
%% =====================================================
-record(time_req, {
  opts :: term()
}).
-record(time_resp, {
  result :: #time{} | #time_error{}
}).

Definieren Sie den Service-Controller in time_controller.erl

%% Das Beispiel zeigt nur den relevanten Code. Durch Einfügen in die Vorlage gen_server kann ein funktionierender Service erstellt werden.

%% Initialisierung von gen_server
init(Args) ->
  %% Verbindung zur Austauschstelle herstellen
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% Verarbeitung des Ereignisses beim Verlust der Verbindung zur Austauschstelle. Dieses Ereignis wird auch gesendet, wenn die Austauschstelle noch nicht gestartet ist.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% Verarbeitung der API
handle_info(#time_req{opts = _Opts}, State) ->
  messaging:response_once(Client, #time_resp{
result = #time{ unixtime = time_utils:unixtime(now()), datetime = time_utils:iso8601_fmt(now())}
  });
  {noreply, State};

%% Beendigung des gen_server
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Client-Code

Um eine Anfrage an den Service zu senden, kann in jedem Teil des Clients die Messaging-Request-API aufgerufen werden:

Messaging-Anfrage: request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) von
    ok -> ok;
    _ -> %% Wiederhol- oder Fehlermeldung
end

In einem verteilten System kann die Konfiguration der Komponenten sehr unterschiedlich sein und zur Zeit der Messaging-Anfrage möglicherweise noch nicht gestartet sein oder der Dienstcontroller ist möglicherweise nicht bereit, die Anfrage zu bearbeiten. Daher müssen wir die Antwort der Messaging überprüfen und den Fall eines Fehlers behandeln.
Nach dem erfolgreichen Versand erhält der Kunde eine Antwort oder eine Fehlermeldung vom Dienst.
Wir behandeln beide Fälle in handle_info:

handle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time{unixtime = Utime}}}, State) ->
  ?debugVal(Utime),
  {noreply, State};

handle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time_error{code = ErrorCode}}}, State) ->
  ?debugVal({error, ErrorCode}),
  {noreply, State};

Anfrage-Chunked Antwort

Es ist besser, die Übertragung großer Nachrichten zu vermeiden. Das hängt von der Reaktionsfähigkeit und stabilen Funktionsweise des gesamten Systems ab. Wenn die Antwort auf eine Anfrage viel Speicher benötigt, ist eine Aufteilung in Teile erforderlich.

Ich werde ein paar Beispiele für solche Fälle anführen:

Komponenten tauschen binäre Daten aus, beispielsweise Dateien. Die Aufteilung der Antwort in kleine Teile hilft, effizient mit Dateien beliebiger Größe zu arbeiten und Überläufe zu vermeiden.
Listen. Zum Beispiel müssen wir alle Einträge aus einer großen Tabelle in der Datenbank auswählen und an eine andere Komponente weiterleiten.

Ich nenne solche Antworten einen Zug. Auf jeden Fall sind 1024 Nachrichten mit je 1 MB besser als eine einzige Nachricht mit 1 GB.

In einem Erlang-Cluster erzielen wir einen zusätzlichen Vorteil ‒ die Belastung des Austauschpunktes und des Netzwerks wird reduziert, da die Antworten sofort an den Empfänger gesendet werden, ohne den Austauschpunkt zu passieren.

Antwort mit Anfrage

Das ist eine eher seltene Modifikation des RPC-Musters zur Erstellung dialogorientierter Systeme.

Ereignisgesteuerte Systeme liefern Daten an Verbraucher, sobald sie bereit sind. Dadurch tendieren die Systeme eher zu einem Push-Modell als zu Pull oder Poll. Diese Eigenschaft verhindert, dass Ressourcen verschwendet werden, indem ständig nach Daten gefragt und darauf gewartet wird.
Das Diagramm zeigt den Prozess der Nachrichtenverbreitung an Verbraucher, die sich auf ein bestimmtes Thema abonniert haben.

Klassische Beispiele für die Verwendung dieser Vorlage sind die Verbreitung des Zustands: der virtuelle Spielwelt in Computerspielen, der Marktdaten an Börsen und nützlicher Informationen in Data Feeds.

Betrachten wir den Code des Abonnenten:

init(_Args) ->
  %% wir abonnieren den Broker, Schlüssel = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% wenn der Broker nicht verfügbar ist, versuchen wir Neuverbindung
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% Verarbeitung eingehender Nachrichten
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% Bei Beendigung des Verbrauchers - Trennung vom Broker
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

Die Quelle kann die Funktion zur Veröffentlichung von Nachrichten an beliebiger Stelle aufrufen:

messaging:publish_message(Exchange, Key, Message).

Exchange ‒ Name des Brokers,
Schlüssel ‒ Routing-Schlüssel
Nachricht ‒ Nutzlast

Durch die Implementierung von pub-sub kann ein Pattern erstellt werden, das sich gut für das Logging eignet. Die Kombination aus Quellen und Verbrauchern kann völlig unterschiedlich sein. Im Bild ist ein Fall mit einem Verbraucher und vielen Quellen dargestellt.

Taskverteilungsmuster

In nahezu jedem Projekt entstehen Aufgaben der verzögerten Verarbeitung, wie die Erstellung von Berichten, die Zustellung von Benachrichtigungen und das Abrufen von Daten aus externen Systemen. Die Kapazität des Systems, das diese Aufgaben ausführt, lässt sich leicht skalieren, indem weitere Bearbeiter hinzugefügt werden. Alles, was wir tun müssen, ist, einen Cluster von Bearbeitern zu bilden und die Aufgaben gleichmäßig zwischen ihnen zu verteilen.

Betrachten wir die auftretenden Situationen am Beispiel von 3 Bearbeitern. Bereits in der Phase der Aufgabenverteilung stellt sich die Frage nach der Fairness der Verteilung und der Überlastung der Bearbeiter. Die Fairness wird durch eine round-robin Verteilung gewährleistet, und um Überlastungen der Bearbeiter zu vermeiden, führen wir eine Einschränkung ein prefetch_limit. In Übergangsmodi prefetch_limit wird kein Bearbeiter alle Aufgaben erhalten.

Messaging verwaltet die Warteschlangen und die Priorität der Verarbeitung. Bearbeiter erhalten die Aufgaben, sobald sie eintreffen. Die Ausführung einer Aufgabe kann erfolgreich abgeschlossen werden oder auch fehlschlagen:

messaging:ack(Tack) ‒ wird im Falle einer erfolgreichen Verarbeitung der Nachricht aufgerufen
messaging:nack(Tack) ‒ wird in allen Notfallsituationen aufgerufen. Nach der Rückgabe der Aufgabe wird Messaging sie an einen anderen Handler übergeben.

Angenommen, beim Verarbeiten von drei Aufgaben trat ein komplexer Fehler auf: Handler 1 ist nach Erhalt der Aufgabe abgestürzt, ohne etwas an den Austauschpunkt zu melden. In diesem Fall wird der Austauschpunkt nach Ablauf des Ack-Timeouts die Aufgabe an einen anderen Handler übergeben. Handler 3 hat aus irgendeinem Grund die Aufgabe abgelehnt und ein Nack gesendet, deswegen wurde die Aufgabe ebenfalls an einen anderen Handler übergeben, der sie erfolgreich ausgeführt hat.

Vorläufiges Fazit

Wir haben die grundlegenden Bausteine verteilter Systeme betrachtet und ein grundlegendes Verständnis ihrer Anwendung in Erlang/Elixir gewonnen.

Durch die Kombination grundlegender Muster können komplexe Paradigmen zur Lösung auftretender Aufgaben entwickelt werden.

Im abschließenden Teil des Zyklus werden wir allgemeine Fragen zur Organisation von Diensten, Routing und Lastverteilung betrachten und über die praktische Seite der Skalierbarkeit und Ausfallsicherheit von Systemen sprechen.

Ende des zweiten Teils.

Foto Marius Christensen
Illustrationen erstellt mit websequencediagrams.com

Quelle: habr.com