Bloques de construcción de aplicaciones distribuidas. Primer enfoque

En el pasado статье Examinamos los fundamentos teóricos de la arquitectura reactiva. Es hora de hablar sobre flujos de datos, formas de implementar sistemas reactivos Erlang/Elixir y patrones de mensajería en ellos:

• Solicitar respuesta
• Respuesta fragmentada de solicitud
• Respuesta con solicitud
• Publicar-suscribir
• Publicación-suscripción invertida
• Distribución de tareas

SOA, MSA y mensajería

SOA y MSA son arquitecturas de sistemas que definen las reglas para construir sistemas, mientras que la mensajería proporciona primitivas para su implementación.

No quiero promover tal o cual arquitectura de sistema. Estoy a favor de utilizar las prácticas más efectivas y útiles para un proyecto y negocio específico. Cualquiera que sea el paradigma que elijamos, es mejor crear bloques de sistema con la vista puesta en el estilo Unix: componentes con conectividad mínima, responsables de entidades individuales. Los métodos API realizan las acciones más simples posibles con entidades.

La mensajería es, como su nombre indica, un intermediario de mensajes. Su objetivo principal es recibir y enviar mensajes. Es responsable de las interfaces para enviar información, la formación de canales lógicos para transmitir información dentro del sistema, el enrutamiento y el equilibrio, así como el manejo de fallas a nivel del sistema.
El mensaje que estamos desarrollando no intenta competir ni reemplazar a Rabbitmq. Sus principales características:

• Distribución.
  Se pueden crear puntos de intercambio en todos los nodos del clúster, lo más cerca posible del código que los utiliza.
• Simplicidad
  Concéntrese en minimizar el código repetitivo y la facilidad de uso.
• El mejor rendimiento.
  No intentamos repetir la funcionalidad de RabbitMQ, sino resaltar solo la capa arquitectónica y de transporte, que encajamos en la OTP de la manera más simple posible, minimizando los costos.
• Flexibilidad.
  Cada servicio puede combinar muchas plantillas de intercambio.
• Resiliencia por diseño.
• Escalabilidad.
  La mensajería crece con la aplicación. A medida que aumenta la carga, puedes mover los puntos de intercambio a máquinas individuales.

Observación En términos de organización del código, los metaproyectos son muy adecuados para sistemas complejos de Erlang/Elixir. Todo el código del proyecto se encuentra en un repositorio: un proyecto general. Al mismo tiempo, los microservicios están aislados al máximo y realizan operaciones simples que son responsables de una entidad separada. Con este enfoque, es fácil mantener la API de todo el sistema, es fácil realizar cambios y es conveniente escribir pruebas unitarias y de integración.

Los componentes del sistema interactúan directamente o a través de un intermediario. Desde la perspectiva de la mensajería, cada servicio tiene varias fases de vida:

• Inicialización del servicio.
  En esta etapa, se configuran y lanzan el proceso y las dependencias que ejecutan el servicio.
• Creación de un punto de intercambio.
  El servicio puede utilizar un punto de intercambio estático especificado en la configuración del nodo o crear puntos de intercambio dinámicamente.
• Registro de servicio.
  Para que el servicio atienda solicitudes, debe estar registrado en el punto de cambio.
• Funcionamiento normal.
  El servicio produce un trabajo útil.
• Cerrar.
  Hay 2 tipos de apagado posibles: normal y de emergencia. Durante el funcionamiento normal, el servicio se desconecta del punto de cambio y se detiene. En situaciones de emergencia, la mensajería ejecuta uno de los scripts de conmutación por error.

Parece bastante complicado, pero el código no da tanto miedo. Se darán ejemplos de código con comentarios en el análisis de plantillas un poco más adelante.

Cambios

El punto de intercambio es un proceso de mensajería que implementa la lógica de interacción con componentes dentro de la plantilla de mensajería. En todos los ejemplos presentados a continuación, los componentes interactúan a través de puntos de intercambio, cuya combinación forma la mensajería.

Patrones de intercambio de mensajes (MEP)

A nivel mundial, los patrones de intercambio se pueden dividir en bidireccionales y unidireccionales. Los primeros implican una respuesta a un mensaje entrante, los segundos no. Un ejemplo clásico de un patrón bidireccional en la arquitectura cliente-servidor es el patrón Solicitud-respuesta. Veamos la plantilla y sus modificaciones.

Solicitud-respuesta o RPC

RPC se utiliza cuando necesitamos recibir una respuesta de otro proceso. Este proceso puede ejecutarse en el mismo nodo o estar ubicado en un continente diferente. A continuación se muestra un diagrama de la interacción entre el cliente y el servidor a través de mensajería.

Dado que la mensajería es completamente asincrónica, para el cliente el intercambio se divide en 2 fases:

1. Enviando una solicitud

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

   Intercambie ‒ nombre único del punto de cambio
   Etiqueta coincidente de respuesta ‒ etiqueta local para procesar la respuesta. Por ejemplo, en el caso de enviar varias solicitudes idénticas pertenecientes a diferentes usuarios.
   SolicitudDefinición - cuerpo de la solicitud
   Proceso de controlador ‒ PID del controlador. Este proceso recibirá una respuesta del servidor.

2. Procesando la respuesta

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

   Carga útil de respuesta - respuesta del servidor.

Para el servidor, el proceso también consta de 2 fases:

1. Inicializando el punto de intercambio
2. Procesamiento de solicitudes recibidas

Ilustremos esta plantilla con código. Digamos que necesitamos implementar un servicio simple que proporcione un único método de hora exacta.

código del servidor

Definamos la API del servicio en 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{}
}).

Definamos el controlador de servicio en 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.

Codigo del cliente

Para enviar una solicitud al servicio, puede llamar a la API de solicitud de mensajería en cualquier parte del cliente:

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

En un sistema distribuido, la configuración de los componentes puede ser muy diferente y, en el momento de la solicitud, es posible que la mensajería aún no comience o que el controlador de servicio no esté listo para atender la solicitud. Por lo tanto, debemos verificar la respuesta del mensaje y manejar el caso de falla.
Después del envío exitoso, el cliente recibirá una respuesta o error del servicio.
Manejemos ambos casos en 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};

Respuesta fragmentada de solicitud

Es mejor evitar enviar mensajes enormes. De esto depende la capacidad de respuesta y el funcionamiento estable de todo el sistema. Si la respuesta a una consulta ocupa mucha memoria, entonces es obligatorio dividirla en partes.

Déjame darte un par de ejemplos de tales casos:

• Los componentes intercambian datos binarios, como archivos. Dividir la respuesta en partes pequeñas le ayuda a trabajar de manera eficiente con archivos de cualquier tamaño y evitar desbordamientos de memoria.
• Listados. Por ejemplo, necesitamos seleccionar todos los registros de una tabla enorme en la base de datos y transferirlos a otro componente.

A estas respuestas las llamo locomotora. En cualquier caso, 1024 mensajes de 1 MB son mejores que un único mensaje de 1 GB.

En el clúster de Erlang, obtenemos un beneficio adicional: reducir la carga en el punto de intercambio y la red, ya que las respuestas se envían inmediatamente al destinatario, sin pasar por el punto de intercambio.

Respuesta con solicitud

Esta es una modificación bastante rara del patrón RPC para crear sistemas de diálogo.

Publicar-suscribir (árbol de distribución de datos)

Los sistemas basados ​​en eventos los entregan a los consumidores tan pronto como los datos están listos. Por lo tanto, los sistemas son más propensos a un modelo push que a un modelo pull o poll. Esta característica le permite evitar el desperdicio de recursos al solicitar y esperar datos constantemente.
La figura muestra el proceso de distribución de un mensaje a los consumidores suscritos a un tema específico.

Ejemplos clásicos del uso de este patrón son la distribución del estado: el mundo del juego en los juegos de computadora, los datos del mercado en las bolsas, la información útil en los flujos de datos.

Veamos el código de suscriptor:

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 fuente puede llamar a la función para publicar un mensaje en cualquier lugar conveniente:

messaging:publish_message(Exchange, Key, Message).

Intercambie - nombre del punto de cambio,
Clave - clave de enrutamiento
Mensaje - carga útil

Publicación-suscripción invertida

Al expandir pub-sub, puede obtener un patrón conveniente para iniciar sesión. El conjunto de fuentes y consumidores puede ser completamente diferente. La figura muestra un caso con un consumidor y múltiples fuentes.

Patrón de distribución de tareas

Casi todos los proyectos implican tareas de procesamiento diferido, como generar informes, entregar notificaciones y recuperar datos de sistemas de terceros. El rendimiento del sistema que realiza estas tareas se puede escalar fácilmente agregando controladores. Todo lo que nos queda es formar un grupo de procesadores y distribuir uniformemente las tareas entre ellos.

Veamos las situaciones que surgen usando el ejemplo de 3 controladores. Incluso en la etapa de distribución de tareas, surge la cuestión de la equidad de la distribución y el exceso de encargados. La distribución por turnos será responsable de la equidad y, para evitar una situación de desbordamiento de controladores, introduciremos una restricción. límite_prefetch. En condiciones transitorias límite_prefetch impedirá que un controlador reciba todas las tareas.

La mensajería gestiona las colas y la prioridad de procesamiento. Los procesadores reciben tareas a medida que llegan. La tarea puede completarse exitosamente o fallar:

• messaging:ack(Tack) - llamado si el mensaje se procesa exitosamente
• messaging:nack(Tack) - llamado en todas las situaciones de emergencia. Una vez que se devuelve la tarea, la mensajería la pasará a otro controlador.

Supongamos que se produjo un fallo complejo durante el procesamiento de tres tareas: el procesador 1, después de recibir la tarea, se bloqueó sin tener tiempo de informar nada al punto de intercambio. En este caso, el punto de intercambio transferirá la tarea a otro controlador después de que haya expirado el tiempo de espera de confirmación. Por alguna razón, el controlador 3 abandonó la tarea y envió nack; como resultado, la tarea también fue transferida a otro controlador que la completó con éxito.

Resultado preliminar

Hemos cubierto los componentes básicos de los sistemas distribuidos y obtuvimos una comprensión básica de su uso en Erlang/Elixir.

Al combinar patrones básicos, se pueden construir paradigmas complejos para resolver problemas emergentes.

En la parte final de la serie, veremos cuestiones generales de organización de servicios, enrutamiento y equilibrio, y también hablaremos sobre el lado práctico de la escalabilidad y la tolerancia a fallas de los sistemas.

Fin de la segunda parte.

Galleria Marius Christensen
Ilustraciones preparadas con websequencediagrams.com

Fuente: habr.com