Bausteine verteilt Anwendungen. Erste Näherung

Letzte Woche Artikel Wir haben die theoretischen Grundlagen der reaktiven Architektur behandelt. Es ist Zeit, über Datenströme, die Implementation von reaktiven Erlang/Elixir-Systemen und die Kommunikationsmuster zu sprechen:

• Request-Response
• Request-Chunked Response
• Response mit Request
• Publish-Subscribe
• Inverted Publish-Subscribe
• Aufgabenverteilung

SOA, MSA und Nachrichtenübermittlung

SOA und MSA sind Systemarchitekturen, die die Regeln zum Aufbau von Systemen definieren, während Messaging Primtivitäten für deren Umsetzung bereitstellt.

Ich möchte keine bestimmte Architektur propagieren. Ich befürworte die Anwendung der für das jeweilige Projekt und Geschäft effektivsten und nützlichsten Praktiken. Welche Paradigmen auch immer wir wählen, es ist besser, systemische Bausteine im Sinne des Unix-Ansatzes zu schaffen: Komponenten mit minimaler Kopplung, die für einzelne Entitäten verantwortlich sind. API-Methoden führen einfachste Aktionen mit diesen Entitäten aus.

Messaging ‒ wie der Name schon sagt ‒ ist ein Nachrichtenbroker. Sein Hauptziel ist es, Nachrichten zu empfangen und zu senden. Es ist verantwortlich für die Schnittstellen zur Informationsübertragung, die Bildung logischer Informationskanäle innerhalb des Systems, das Routing und die Lastverteilung sowie die Fehlerbehandlung auf Systemebene.
Das entwickelte Messaging versucht nicht, mit RabbitMQ zu konkurrieren oder es zu ersetzen. Seine Hauptmerkmale:

• Verteilung.
  Austauschpunkte können auf allen Knoten des Clusters, möglichst nah am Code, der sie verwendet, erstellt werden.
• Einfachheit.
  Fokus auf die Minimierung von Boilerplate-Code und Benutzerfreundlichkeit.
• Bessere Leistung.
  Wir versuchen nicht, die Funktionalität von RabbitMQ zu wiederholen, sondern heben lediglich die architektonische und transporttechnische Schicht hervor, die wir sehr einfach in OTP integrieren, um die Kosten zu minimieren.
• Flexibilität.
  Jeder Dienst kann viele Austauschmuster kombinieren.
• Fehlertoleranz, die in das Design integriert ist.
• Skalierbarkeit.
  Messaging wächst mit der Anwendung. Mit zunehmender Last können Austauschpunkte auf separate Maschinen ausgelagert werden.

Anmerkung. Aus Sicht der Code-Organisation sind Metaprojekte für komplexe Systeme in Erlang/Elixir gut geeignet. Der gesamte Code des Projekts befindet sich in einem einzigen Repository – dem übergeordneten Projekt. Dabei sind die Mikrodienste maximal isoliert und führen einfache Operationen aus, die für eine einzelne Entität zuständig sind. Mit diesem Ansatz ist es einfach, die API des gesamten Systems zu pflegen, Änderungen vorzunehmen und Unit- sowie Integrationstests zu schreiben.

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

• Initialisierung des Dienstes.
  In diesem Stadium erfolgt die Konfiguration und der Start des ausführenden Dienstprozesses und dessen Abhängigkeiten.
• Einrichtung eines Austauschpunkts.
  Der Dienst kann einen statischen Austauschpunkt verwenden, der in der Konfiguration des Knotens festgelegt ist, oder er kann Austauschpunkte dynamisch erstellen.
• Registrierung des Dienstes.
  Damit der Dienst Anfragen bearbeiten kann, muss er am Austauschpunkt registriert werden.
• Regelmäßiger Betrieb.
  Der Dienst führt nützliche Arbeiten aus.
• Beendigung der Arbeit.
  Es gibt zwei Arten der Beendigung: regulär und abrupt. Bei regulärer Beendigung wird der Dienst vom Austauschpunkt getrennt und gestoppt. In Notfällen führt das Messaging eines der Szenarien zur Fehlerbehandlung aus.

Es sieht ziemlich kompliziert aus, aber im Code ist nicht alles so schlimm. Codebeispiele mit Kommentaren werden später bei der Analyse der Muster vorgestellt.

Exchanges

Der Austauschpunkt ist ein Messaging-Prozess, der die Logik der Interaktion mit den Komponenten im Rahmen des Messaging-Mustertyps umsetzt. In allen unten dargestellten Beispielen interagieren die Komponenten über Austauschpunkte, deren Kombination das Messaging bildet.

Message Exchange Patterns (MEPs)

Global können die Austauschmuster in bidirektionale und unidirektionale unterteilt werden. Erstere setzen eine Antwort auf die eingegangene Nachricht voraus, letztere nicht. Ein klassisches Beispiel für ein bidirektionales Muster in einer Client-Server-Architektur ist das Request-Response-Muster. Lassen Sie uns das Muster und seine Modifikationen untersuchen.

Request–Response 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. Unten ist das Schema der Interaktion zwischen dem Client und Server über Messaging.

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

1. Anfrage senden

   messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).

   Austausch ‒ einzigartiger Name des Austauschpunkts
   ResponseMatchingTag ‒ lokales Tag zur Verarbeitung der Antwort. Zum Beispiel im Fall der Sendung mehrerer identischer Anfragen, die verschiedenen Benutzern gehören.
   RequestDefinition ‒ Anfragekörper
   HandlerProcess ‒ PID des Handlers. Dieser Prozess erhält die Antwort vom Server.

2. Antwortverarbeitung

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

   ResponsePayload ‒ Serverantwort.

Für den Server besteht der Prozess ebenfalls aus 2 Phasen:

1. Initialisierung des Austauschpunkts
2. Verarbeitung der eingehenden Anfragen

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

Servercode

Lass uns die API-Definition 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 wir den Dienst-Controller in time_controller.erl

%% Im Beispiel wird nur der wesentliche Code gezeigt. Indem wir ihn in das gen_server-Muster einfügen, können wir einen funktionierenden Dienst erhalten.

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

%% Verarbeitung des Ereignisses eines Verbindungsverlusts zum Austauschpunkt. Dasselbe Ereignis tritt auf, wenn der Austauschpunkt 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 der gen_server-Arbeit
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Client-Code

Um eine Anfrage an den Dienst zu senden, kann an beliebiger Stelle im Client die messaging request API aufgerufen werden:

case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% Wiederholungs- oder Fehlerlogik
end

In einem verteilten System kann die Konfiguration der Komponenten unterschiedlich sein und zum Zeitpunkt der Anfrage kann die messaging noch nicht gestartet sein oder der Dienst-Controller 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 erfolgreicher Sendung erhält der Client eine Antwort oder einen Fehler vom Dienst.
Behandeln wir 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};

Request-Chunked Response

Es ist besser, keine riesigen Nachrichten zu übermitteln. Das beeinflusst die Reaktionsfähigkeit und die stabile Funktionsweise des gesamten Systems. Wenn die Antwort auf eine Anfrage viel Speicher benötigt, ist eine Aufteilung in Teile erforderlich.

Hier sind ein paar Beispiele für solche Fälle:

• Komponenten tauschen Binärdaten, wie Dateien, aus. Die Aufteilung der Antwort in kleinere Teile hilft, effizient mit Dateien jeder Größe zu arbeiten und ein Speicherüberlauf zu vermeiden.
• Listen. Zum Beispiel müssen wir alle Datensätze aus einer riesigen Tabelle in der Datenbank auswählen und an eine andere Komponente übermitteln.

Ich nenne solche Antworten einen Zug. In jedem Fall sind 1024 Nachrichten zu je 1 MB besser als eine einzige Nachricht von 1 GB.

In einem Erlang-Cluster erzielen wir einen zusätzlichen Vorteil – die Reduzierung der Belastung an der Austauschstelle und im Netzwerk, da die Antworten sofort an den Empfänger gesendet werden, ohne die Austauschstelle zu passieren.

Response mit Request

Dies ist eine recht seltene Modifikation des RPC-Musters zum Aufbau dialogorientierter Systeme.

Publish-Subscribe (Datenverteilung Baum)

Ereignisgesteuerte Systeme liefern Daten an Verbraucher, sobald sie bereit sind. Dadurch tendieren die Systeme eher zu einem Push-Modell als zu Pull- oder Poll-Modellen. Diese Eigenschaft verhindert, dass Ressourcen unnötig verschwendet werden, indem ständig Daten angefordert und erwartet werden.
Die Abbildung zeigt den Prozess der Verbreitung der Nachricht an die Verbraucher, die sich für ein bestimmtes Thema angemeldet haben.

Klassische Beispiele für die Verwendung dieses Musters sind die Verbreitung von Zuständen: Spielwelten in Computerspielen, Marktdaten an Börsen, nützliche Informationen in Datenfeeds.

Betrachten wir den Code des Abonnenten:

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

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% wenn die Austauschstelle nicht verfügbar ist, versuchen wir, uns erneut zu verbinden
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

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

%% beim Stoppen des Verbrauchers trennen wir uns von der Austauschstelle
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

Die Quelle kann die Funktion zum Veröffentlichen von Nachrichten an beliebiger Stelle aufrufen:

messaging:publish_message(Exchange, Key, Message).

Austausch ‒ Name des Austauschpunkts,
Schlüssel ‒ Routing-Schlüssel
Nachricht ‒ Payload

Inverted Publish-Subscribe

Durch die Einführung von Pub-Sub kann ein Muster erzielt werden, das sich zum Logging eignet. Die Sammlung von Quellen und Verbrauchern kann vollkommen unterschiedlich sein. Im Bild ist ein Fall mit einem Verbraucher und mehreren Quellen dargestellt.

Task-Verteilungsmuster

In fast jedem Projekt gibt es Aufgaben zur verzögerten Verarbeitung, wie zum Beispiel das Erstellen von Berichten, das Versenden von Benachrichtigungen und das Abrufen von Daten aus externen Systemen. Die Kapazität des Systems, das diese Aufgaben ausführt, kann leicht durch das Hinzufügen von Handlern skaliert werden. Alles, was wir tun müssen, ist, einen Cluster von Handlern zu bilden und die Aufgaben gleichmäßig zwischen ihnen zu verteilen.

Betrachten wir die auftretenden Situationen am Beispiel von 3 Handlern. Bereits in der Phase der Aufgabenverteilung stellt sich die Frage der Fairness der Verteilung und der Überlastung der Handler. Für die Fairness wird eine Round-Robin-Verteilung sorgen, um Überlastung der Handler zu vermeiden, führen wir eine Begrenzung ein prefetch_limit. In Übergangsmodi prefetch_limit wird es einem Handler nicht ermöglichen, alle Aufgaben zu erhalten.

Messaging verwaltet Warteschlangen und Prioritäten bei der Verarbeitung. Handler erhalten Aufgaben, sobald sie anfallen. Die Ausführung einer Aufgabe kann erfolgreich abgeschlossen werden oder fehlschlagen:

• messaging:ack(Tack) ‒ wird im Falle einer erfolgreichen Verarbeitung der Nachricht aufgerufen
• messaging:nack(Tack) ‒ wird in allen Ausnahmefällen aufgerufen. Nach der Rückgabe der Aufgabe wird messaging sie an einen anderen Handler weitergeben.

Angenommen, während der Verarbeitung von drei Aufgaben tritt ein komplexer Fehler auf: Handler 1 fällt nach dem Erhalt der Aufgabe aus, ohne der Austauschstelle etwas mitgeteilt zu haben. In diesem Fall wird die Austauschstelle nach Ablauf der Ack-Timeout die Aufgabe an einen anderen Handler weitergeben. Handler 3 hat aus irgendeinem Grund die Aufgabe abgelehnt und nack gesendet, sodass die Aufgabe ebenfalls an einen anderen Handler übergeben wurde, der sie erfolgreich ausgeführt hat.

Vorläufiges Ergebnis

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

Durch die Kombination grundlegender Muster können komplexe Paradigmen zum Lösen auftretender Aufgaben aufgebaut werden.

Im abschließenden Teil des Zyklus werden wir allgemeine Fragen zur Organisation von Diensten, zur Routenführung und zum Lastenausgleich sowie zur praktischen Seite der Skalierbarkeit und Ausfallsicherheit von Systemen erörtern.

Ende des zweiten Teils.

Foto Marius Christensen
Illustrationen wurden mit Hilfe von websequencediagrams.com erstellt.

Quelle: habr.com