Bausteine ​​verteilter Anwendungen. Erste Ansatz

In der Vergangenheit Artikel Wir haben die theoretischen Grundlagen der reaktiven Architektur analysiert. Es ist an der Zeit, über Datenflüsse, Möglichkeiten zur Implementierung reaktiver Erlang/Elixir-Systeme und darin enthaltene Nachrichtenmuster zu sprechen:

• Anfrage-Antwort
• Anfrage-Chunked-Antwort
• Antwort mit Anfrage
• Veröffentlichen-Abonnieren
• Invertiert „Publish Subscribe“.
• Aufgabenverteilung

SOA, MSA und Messaging

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

Ich möchte diese oder jene Systemarchitektur nicht propagieren. Ich bin für die Anwendung der effektivsten und nützlichsten Praktiken für ein bestimmtes Projekt und Unternehmen. Für welches Paradigma wir uns auch entscheiden, es ist besser, Systemblöcke mit Blick auf die Unix-Methode zu erstellen: Komponenten mit minimaler Konnektivität, die für einzelne Einheiten verantwortlich sind. API-Methoden führen die einfachsten Aktionen mit Entitäten aus.

Messaging ist, wie der Name schon sagt, ein Nachrichtenbroker. Sein Hauptzweck besteht darin, Nachrichten zu empfangen und zu senden. Es ist verantwortlich für die Schnittstellen zum Senden von Informationen, die Bildung logischer Kanäle zur Übertragung von Informationen innerhalb des Systems, Routing und Balancing 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 Clusterknoten so nah wie möglich an dem Code erstellt werden, der sie verwendet.
• Einfachheit.
  Konzentrieren Sie sich auf die Minimierung des Boilerplate-Codes und die Benutzerfreundlichkeit.
• Der beste Auftritt.
  Wir versuchen nicht, die Funktionalität von Rabbitmq zu wiederholen, sondern wählen nur die Architektur- und Transportschicht aus, die wir so einfach wie möglich in OTP integrieren und so die Kosten minimieren.
• Flexibilität.
  Jeder Dienst kann viele Austauschvorlagen kombinieren.
• Ausfallsicherheit durch Design.
• Skalierbarkeit.
  Die Nachrichtenübermittlung wächst mit der App. Bei steigender Auslastung können Sie die Austauschpunkte auf separate Maschinen verlegen.

ANMERKUNG. Im Hinblick auf die Codeorganisation eignen sich Metaprojekte gut für komplexe Erlang/Elixir-Systeme. Der gesamte Projektcode befindet sich in einem Repository – einem Dachprojekt. Gleichzeitig sind Microservices so weit wie möglich isoliert und führen einfache Vorgänge aus, die für eine separate Entität verantwortlich sind. Mit diesem Ansatz ist es einfach, die API des gesamten Systems zu warten, es ist einfach, Änderungen vorzunehmen, und es ist bequem, Unit- und Integrationstests zu schreiben.

Systemkomponenten interagieren direkt oder über einen Broker. Aus Sicht der Nachrichtenübermittlung hat jeder Dienst mehrere Lebensphasen:

• Dienstinitialisierung.
  In dieser Phase erfolgt die Konfiguration und der Start des Prozesses, der den Dienst und die Abhängigkeiten ausführt.
• Erstellen Sie einen Austauschpunkt.
  Der Dienst kann einen in der Hostkonfiguration angegebenen statischen Austauschpunkt verwenden oder Austauschpunkte dynamisch erstellen.
• Serviceregistrierung.
  Damit der Dienst Anfragen bedienen kann, muss er am Austauschpunkt registriert sein.
• Normale Funktion.
  Der Dienst leistet nützliche Arbeit.
• Abschluss der Arbeit.
  Es sind zwei Arten der Abschaltung möglich: Normal und Notfall. Im Normalbetrieb wird der Dienst vom Vermittlungspunkt getrennt und gestoppt. In Notfallsituationen führt Messaging eines der Failover-Skripte aus.

Es sieht ziemlich kompliziert aus, aber der Code ist nicht so gruselig. Codebeispiele mit Kommentaren werden etwas später in der Analyse von Vorlagen angegeben.

Warenumtausch

Ein Austauschpunkt ist ein Messaging-Prozess, der die Logik der Interaktion mit Komponenten innerhalb der Messaging-Vorlage implementiert. In allen folgenden Beispielen interagieren die Komponenten über Austauschpunkte, deren Kombination eine Nachrichtenübermittlung bildet.

Nachrichtenaustauschmuster (MEPs)

Global gesehen können Austauschmuster in zweiseitige und einseitige unterteilt werden. Ersteres impliziert eine Antwort auf die eingehende Nachricht, letzteres nicht. Ein klassisches Beispiel für ein bidirektionales Muster in einer Client-Server-Architektur ist das Request-Response-Muster. Berücksichtigen Sie die Vorlage und ihre Änderungen.

Anfrage-Antwort oder RPC

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

Da die Nachrichtenübermittlung vollständig asynchron erfolgt, ist der Austausch für den Client in zwei Phasen unterteilt:

1. Senden einer Anfrage

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

   Austausch ‒ eindeutiger Name des Austauschpunkts
   ResponseMatchingTag ‒ lokales Label zur Verarbeitung der Antwort. Beispielsweise beim Versenden mehrerer identischer Anfragen verschiedener Benutzer.
   Anforderungsdefinition ‒ Anfragetext
   HandlerProcess ‒ PID des Handlers. Dieser Vorgang erhält eine 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 Austauschpunktes
2. Bearbeitung eingehender Anfragen

Lassen Sie uns diese Vorlage mit Code veranschaulichen. Nehmen wir an, wir müssen einen einfachen Dienst implementieren, der eine einzige exakte Zeitmethode bereitstellt.

Servercode

Verschieben wir die Service-API-Definition nach api.hrl:

%% =====================================================
%%  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

%% В примере показан только значимый код. Вставив его в шаблон gen_server можно получить рабочий сервис.

%% инициализация gen_server
init(Args) ->
  %% подключение к точке обмена
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% обработка события потери связи с точкой обмена. Это же событие приходит, если точка обмена еще не запустилась.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% обработка 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};

%% завершение работы gen_server
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Kundencode

Um eine Anfrage an einen Dienst zu senden, können Sie die Messaging-Anfrage-API an einer beliebigen Stelle auf dem Client aufrufen:

case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% repeat or fail logic
end

In einem verteilten System kann die Konfiguration der Komponenten sehr unterschiedlich sein und zum Zeitpunkt der Anforderung kann es sein, dass die Nachrichtenübermittlung noch nicht beginnt oder der Dienstcontroller nicht bereit ist, die Anforderung zu bearbeiten. Daher müssen wir die Nachrichtenantwort überprüfen und den Fehlerfall behandeln.
Nach erfolgreichem Senden an den Client erhält der Dienst eine Antwort oder einen Fehler.
Lassen Sie uns beide Fälle in handle_info behandeln:

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 am besten, das Versenden großer Nachrichten zu vermeiden. Davon hängt die Reaktionsfähigkeit und der stabile Betrieb des Gesamtsystems ab. Wenn die Antwort auf eine Anfrage viel Speicher beansprucht, ist die Aufteilung in Teile zwingend erforderlich.

Hier einige Beispiele für solche Fälle:

• Komponenten tauschen Binärdaten, beispielsweise Dateien, aus. Das Aufteilen der Antwort in kleine Teile hilft, effizient mit Dateien jeder Größe zu arbeiten und keinen Speicherüberlauf zu bemerken.
• Einträge. Beispielsweise müssen wir alle Datensätze aus einer riesigen Tabelle in der Datenbank auswählen und an eine andere Komponente übergeben.

Ich nenne solche Reaktionen eine Lokomotive. In jedem Fall sind 1024 1-MB-Nachrichten besser als eine einzelne 1-GB-Nachricht.

Im Erlang-Cluster erhalten wir einen zusätzlichen Vorteil: Die Belastung des Austauschpunkts und des Netzwerks wird reduziert, da Antworten unter Umgehung des Austauschpunkts sofort an den Empfänger gesendet werden.

Antwort mit Anfrage

Dies ist eine eher seltene Modifikation des RPC-Musters zum Aufbau von Konversationssystemen.

Publish-Subscribe (Datenverteilungsbaum)

Ereignisgesteuerte Systeme liefern sie an Verbraucher, sobald die Daten bereit sind. Daher sind Systeme anfälliger für ein Push-Modell als für ein Pull- oder Poll-Modell. Mit dieser Funktion können Sie die Verschwendung von Ressourcen durch ständiges Anfordern und Warten auf Daten vermeiden.
Die Abbildung zeigt den Prozess der Verteilung einer Nachricht an Verbraucher, die ein bestimmtes Thema abonniert haben.

Klassische Beispiele für die Verwendung dieses Musters sind die Zustandsverteilung: die Spielwelt in Computerspielen, Marktdaten an Börsen, nützliche Informationen in Datenfeeds.

Betrachten Sie den Abonnentencode:

init(_Args) ->
  %% подписываемся на обменник, ключ = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% если точка обмена недоступна, то пытаемся переподключиться
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% обрабатываем пришедшие сообщения
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% при остановке потребителя - отключаемся от точки обмена
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

Die Quelle kann die Veröffentlichungsfunktion der Nachricht an jedem geeigneten Ort aufrufen:

messaging:publish_message(Exchange, Key, Message).

Austausch - Name der Wechselstelle,
Wesentliche ‒ Routing-Schlüssel
Nachricht - Nutzlast

Invertiert „Publish Subscribe“.

Durch die Bereitstellung von Pub-Sub können Sie ein Muster erhalten, das für die Protokollierung geeignet ist. Die Menge der Quellen und Verbraucher kann völlig unterschiedlich sein. Die Abbildung zeigt einen Fall mit einem Verbraucher und vielen Quellen.

Aufgabenverteilungsmuster

In fast jedem Projekt gibt es Aufgaben mit verzögerter Verarbeitung, wie z. B. das Erstellen von Berichten, das Versenden von Benachrichtigungen und das Empfangen von Daten von Drittsystemen. Der Durchsatz eines Systems, das diese Aufgaben ausführt, lässt sich durch das Hinzufügen von Prozessoren leicht skalieren. Es bleibt uns nur noch, ein Cluster von Prozessoren zu bilden und die Aufgaben gleichmäßig zwischen ihnen zu verteilen.

Schauen wir uns die auftretenden Situationen am Beispiel von 3 Handlern an. Bereits in der Phase der Aufgabenverteilung stellt sich die Frage nach der Verteilungsgerechtigkeit und der Überlastung der Bearbeiter. Für die Fairness ist die Round-Robin-Verteilung verantwortlich. Um eine Überlastung der Handler zu vermeiden, werden wir eine Einschränkung einführen prefetch_limit. In Übergangsmodi prefetch_limit lässt nicht zu, dass ein Handler alle Aufgaben empfängt.

Messaging verwaltet Warteschlangen und Verarbeitungsprioritäten. Bearbeiter erhalten Aufgaben, sobald sie eintreffen. Die Aufgabe kann erfolgreich abgeschlossen werden oder fehlschlagen:

• messaging:ack(Tack) ‒ wird bei erfolgreicher Verarbeitung der Nachricht aufgerufen
• messaging:nack(Tack) - in allen Notsituationen angerufen. Sobald die Aufgabe zurückgegeben wurde, wird sie vom Messaging an einen anderen Handler weitergeleitet.

Nehmen wir an, dass bei der Verarbeitung von drei Aufgaben ein komplexer Fehler aufgetreten ist: Handler 1 stürzte nach Erhalt der Aufgabe ab, ohne Zeit zu haben, dem Austauschpunkt etwas zu melden. In diesem Fall übergibt der Exchange Point den Auftrag nach Ablauf des Ack-Timeouts an einen anderen Handler. Handler 3 hat die Aufgabe aus irgendeinem Grund abgebrochen und einen Nack gesendet. Infolgedessen wurde die Aufgabe auch an einen anderen Handler übergeben, der sie erfolgreich abgeschlossen hat.

Vorläufiges Ergebnis

Wir haben die Grundbausteine ​​verteilter Systeme aufgeschlüsselt und ein grundlegendes Verständnis für deren Verwendung in Erlang/Elixir gewonnen.

Durch die Kombination grundlegender Vorlagen können komplexe Paradigmen zur Lösung aufkommender Probleme erstellt werden.

Im letzten Teil des Zyklus werden wir uns mit allgemeinen Fragen der Organisation von Diensten, Routing und Balancing befassen und auch über die praktische Seite der Skalierbarkeit und Fehlertoleranz von Systemen sprechen.

Ende des zweiten Teils.

Foto Marius Christensen
Illustrationen mit freundlicher Genehmigung von websequencediagrams.com

Source: habr.com