Éléments de base des applications distribuées. Première approche

À la fin article Nous avons examiné les fondements théoriques de l'architecture réactive. Il est temps de parler des flux de données, des moyens d'implémenter des systèmes Erlang/Elixir réactifs et des modèles de messagerie dans ceux-ci :

• Demande de réponse
• Réponse fragmentée à la demande
• Réponse avec demande
• Publier-s'abonner
• Publication-abonnement inversé
• Répartition des tâches

SOA, MSA et messagerie

SOA, MSA sont des architectures système qui définissent les règles de construction des systèmes, tandis que la messagerie fournit des primitives pour leur mise en œuvre.

Je ne veux pas promouvoir telle ou telle architecture système. Je suis favorable à l'utilisation des pratiques les plus efficaces et les plus utiles pour un projet et une entreprise spécifiques. Quel que soit le paradigme que nous choisissons, il est préférable de créer des blocs système en gardant un œil sur la manière Unix : des composants avec une connectivité minimale, responsables d'entités individuelles. Les méthodes API effectuent les actions les plus simples possibles avec les entités.

La messagerie est, comme son nom l'indique, un courtier de messages. Son objectif principal est de recevoir et d'envoyer des messages. Il est responsable des interfaces d'envoi des informations, de la formation de canaux logiques pour la transmission des informations au sein du système, du routage et de l'équilibrage, ainsi que de la gestion des pannes au niveau du système.
La messagerie que nous développons n’essaie pas de rivaliser ou de remplacer RabbitMQ. Ses principales caractéristiques :

• Distribution.
  Des points d'échange peuvent être créés sur tous les nœuds du cluster, au plus près du code qui les utilise.
• Simplicité.
  Concentrez-vous sur la réduction du code passe-partout et la facilité d'utilisation.
• La meilleure performance.
  Nous n'essayons pas de répéter les fonctionnalités de RabbitMQ, mais de souligner uniquement la couche architecturale et de transport, que nous intégrons le plus simplement possible dans l'OTP, en minimisant les coûts.
• La flexibilité.
  Chaque service peut combiner de nombreux modèles d'échange.
• La résilience dès la conception.
• Évolutivité.
  La messagerie évolue avec l'application. À mesure que la charge augmente, vous pouvez déplacer les points d'échange vers des machines individuelles.

REMARQUE En termes d'organisation du code, les méta-projets sont bien adaptés aux systèmes Erlang/Elixir complexes. Tout le code du projet se trouve dans un seul référentiel : un projet parapluie. Dans le même temps, les microservices sont isolés au maximum et effectuent des opérations simples responsables d'une entité distincte. Avec cette approche, il est facile de maintenir l'API de l'ensemble du système, il est facile d'apporter des modifications et il est pratique d'écrire des tests unitaires et d'intégration.

Les composants du système interagissent directement ou via un courtier. Du point de vue de la messagerie, chaque service comporte plusieurs phases de vie :

• Initialisation des services.
  A ce stade, le processus et les dépendances exécutant le service sont configurés et lancés.
• Création d'un point d'échange.
  Le service peut utiliser un point d'échange statique spécifié dans la configuration du nœud ou créer des points d'échange de manière dynamique.
• Inscription aux services.
  Pour que le service puisse répondre aux demandes, il doit être enregistré au point d'échange.
• Fonctionnement normal.
  Le service produit un travail utile.
• Fermer.
  Il existe 2 types d'arrêt possibles : normal et d'urgence. En fonctionnement normal, le service est déconnecté du point d'échange et s'arrête. Dans les situations d'urgence, la messagerie exécute l'un des scripts de basculement.

Cela semble assez compliqué, mais le code n’est pas si effrayant. Des exemples de code avec commentaires seront donnés dans l'analyse des modèles un peu plus tard.

Échanges

Le point d'échange est un processus de messagerie qui implémente la logique d'interaction avec les composants au sein du modèle de messagerie. Dans tous les exemples présentés ci-dessous, les composants interagissent via des points d'échange dont la combinaison forme une messagerie.

Modèles d'échange de messages (MEP)

À l’échelle mondiale, les modèles d’échange peuvent être divisés en deux et unidirectionnels. Les premiers impliquent une réponse à un message entrant, les seconds non. Un exemple classique de modèle bidirectionnel dans l’architecture client-serveur est le modèle requête-réponse. Regardons le modèle et ses modifications.

Requête-réponse ou RPC

RPC est utilisé lorsque nous devons recevoir une réponse d'un autre processus. Ce processus peut s'exécuter sur le même nœud ou être situé sur un continent différent. Vous trouverez ci-dessous un schéma de l'interaction entre le client et le serveur via la messagerie.

La messagerie étant totalement asynchrone, pour le client l'échange se divise en 2 phases :

1. Envoi d'une demande

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

   Échange ‒ nom unique du point d'échange
   ResponseMatchingTag ‒ étiquette locale pour le traitement de la réponse. Par exemple, dans le cas de l’envoi de plusieurs requêtes identiques appartenant à des utilisateurs différents.
   Définition de la demande - corps de la demande
   Processus de gestionnaire – PID du gestionnaire. Ce processus recevra une réponse du serveur.

2. Traitement de la réponse

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

   Charge utile de réponse - réponse du serveur.

Pour le serveur, le processus se compose également de 2 phases :

1. Initialisation du point d'échange
2. Traitement des demandes reçues

Illustrons ce modèle avec du code. Disons que nous devons implémenter un service simple qui fournit une seule méthode d'heure exacte.

Code du serveur

Définissons l'API du service dans 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{}
}).

Définissons le contrôleur de service dans 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.

Code client

Afin d'envoyer une requête au service, vous pouvez appeler l'API de requête de messagerie n'importe où dans le client :

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

Dans un système distribué, la configuration des composants peut être très différente et au moment de la requête, la messagerie peut ne pas encore démarrer ou le contrôleur de service ne sera pas prêt à répondre à la requête. Par conséquent, nous devons vérifier la réponse de la messagerie et gérer le cas d’échec.
Après un envoi réussi, le client recevra une réponse ou une erreur du service.
Traitons les deux cas dans 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};

Réponse fragmentée à la demande

Il est préférable d'éviter d'envoyer des messages énormes. La réactivité et le fonctionnement stable de l'ensemble du système en dépendent. Si la réponse à une requête occupe beaucoup de mémoire, il est obligatoire de la diviser en parties.

Permettez-moi de vous donner quelques exemples de tels cas :

• Les composants échangent des données binaires, telles que des fichiers. Diviser la réponse en petites parties vous aide à travailler efficacement avec des fichiers de toute taille et à éviter les débordements de mémoire.
• Annonces. Par exemple, nous devons sélectionner tous les enregistrements d'une immense table de la base de données et les transférer vers un autre composant.

J’appelle ces réponses une locomotive. Dans tous les cas, 1024 messages de 1 Mo valent mieux qu’un seul message de 1 Go.

Dans le cluster Erlang, nous bénéficions d'un avantage supplémentaire : réduire la charge sur le point d'échange et le réseau, puisque les réponses sont immédiatement envoyées au destinataire, en contournant le point d'échange.

Réponse avec demande

Il s'agit d'une modification assez rare du modèle RPC pour la création de systèmes de dialogue.

Publication-abonnement (arborescence de distribution de données)

Les systèmes événementiels les fournissent aux consommateurs dès que les données sont prêtes. Ainsi, les systèmes sont plus enclins à un modèle push qu’à un modèle pull ou poll. Cette fonctionnalité vous permet d'éviter de gaspiller des ressources en demandant et en attendant constamment des données.
La figure montre le processus de distribution d'un message aux consommateurs abonnés à un sujet spécifique.

Des exemples classiques d'utilisation de ce modèle sont la répartition de l'état : le monde du jeu dans les jeux informatiques, les données de marché sur les bourses, les informations utiles dans les flux de données.

Regardons le code d'abonné :

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.

La source peut appeler la fonction pour publier un message à n'importe quel endroit qui lui convient :

messaging:publish_message(Exchange, Key, Message).

Échange - nom du point d'échange,
ACTIVITES - clé de routage
Message - charge utile

Publication-abonnement inversé

En développant pub-sub, vous pouvez obtenir un modèle pratique pour la journalisation. L'ensemble des sources et des consommateurs peut être complètement différent. La figure montre un cas avec un consommateur et plusieurs sources.

Modèle de répartition des tâches

Presque tous les projets impliquent des tâches de traitement différées, telles que la génération de rapports, l'envoi de notifications et la récupération de données à partir de systèmes tiers. Le débit du système effectuant ces tâches peut être facilement adapté en ajoutant des gestionnaires. Il ne nous reste plus qu'à former un cluster de processeurs et à répartir équitablement les tâches entre eux.

Regardons les situations qui se présentent en prenant l'exemple de 3 handlers. Même au stade de la répartition des tâches, la question de l'équité de la répartition et du débordement de gestionnaires se pose. La distribution à tour de rôle sera responsable de l'équité, et pour éviter une situation de débordement de gestionnaires, nous introduirons une restriction prefetch_limit. En conditions transitoires prefetch_limit empêchera un gestionnaire de recevoir toutes les tâches.

La messagerie gère les files d'attente et la priorité de traitement. Les processeurs reçoivent les tâches au fur et à mesure de leur arrivée. La tâche peut se terminer avec succès ou échouer :

• messaging:ack(Tack) - appelé si le message est traité avec succès
• messaging:nack(Tack) - appelé dans toutes les situations d'urgence. Une fois la tâche renvoyée, la messagerie la transmettra à un autre gestionnaire.

Supposons qu'une panne complexe se produise lors du traitement de trois tâches : le processeur 1, après avoir reçu la tâche, plante sans avoir le temps de signaler quoi que ce soit au point d'échange. Dans ce cas, le point d'échange transférera la tâche vers un autre gestionnaire après l'expiration du délai d'attente d'accusé de réception. Pour une raison quelconque, le gestionnaire 3 a abandonné la tâche et a envoyé nack ; en conséquence, la tâche a également été transférée à un autre gestionnaire qui l'a terminée avec succès.

Résultat préliminaire

Nous avons couvert les éléments de base des systèmes distribués et acquis une compréhension de base de leur utilisation dans Erlang/Elixir.

En combinant des modèles de base, vous pouvez créer des paradigmes complexes pour résoudre des problèmes émergents.

Dans la dernière partie de la série, nous examinerons les questions générales d'organisation des services, de routage et d'équilibrage, et parlerons également du côté pratique de l'évolutivité et de la tolérance aux pannes des systèmes.

Fin de la deuxième partie.

photo Marius Christensen
Illustrations préparées à l’aide de websequencediagrams.com

Source: habr.com