🥇Bloques de construción de aplicacións distribuídas. Primeira aproximación

No último Artigo Examinamos os fundamentos teóricos da arquitectura reactiva. É hora de falar de fluxos de datos, formas de implementar sistemas reactivos Erlang/Elixir e patróns de mensaxería neles:

Solicitude-resposta
Solicitude de resposta fragmentada
Resposta con Solicitude
Publicar-subscribir
Publicación-subscrición invertida
Distribución de tarefas

SOA, MSA e mensaxería

SOA, MSA son arquitecturas de sistemas que definen as regras para construír sistemas, mentres que a mensaxería proporciona primitivas para a súa implementación.

Non quero promover esta ou aquela arquitectura do sistema. Estou a favor de utilizar as prácticas máis eficaces e útiles para un proxecto e negocio específicos. Sexa cal sexa o paradigma que elixamos, é mellor crear bloques de sistema coa vista posta no camiño Unix: compoñentes cunha conectividade mínima, responsables de entidades individuais. Os métodos API realizan as accións máis sinxelas posibles con entidades.

A mensaxería é, como o nome indica, un corredor de mensaxes. A súa finalidade principal é recibir e enviar mensaxes. É responsable das interfaces para o envío de información, a formación de canles lóxicos para transmitir información dentro do sistema, o enrutamento e o equilibrado, así como o tratamento de avarías a nivel do sistema.
A mensaxe que estamos a desenvolver non tenta competir nin substituír rabbitmq. As súas principais características:

Distribución.
Pódense crear puntos de intercambio en todos os nodos do clúster, o máis preto posible do código que os utiliza.
Sinxeleza.
Concéntrase en minimizar o código estándar e a facilidade de uso.
Mellor rendemento.
Non intentamos repetir a funcionalidade de rabbitmq, senón destacar só a capa arquitectónica e de transporte, que encaixamos na OTP da forma máis sinxela posible, minimizando os custos.
Flexibilidade.
Cada servizo pode combinar moitos modelos de intercambio.
Resiliencia por deseño.
Escalabilidade.
A mensaxería crece coa aplicación. A medida que aumenta a carga, pode mover os puntos de intercambio a máquinas individuais.

Observación En termos de organización do código, os metaproxectos son moi adecuados para sistemas complexos Erlang/Elixir. Todo o código do proxecto está situado nun repositorio: un proxecto paraugas. Ao mesmo tempo, os microservizos están illados ao máximo e realizan operacións sinxelas que son responsables dunha entidade separada. Con este enfoque, é fácil manter a API de todo o sistema, é fácil facer cambios, é conveniente escribir probas unitarias e de integración.

Os compoñentes do sistema interactúan directamente ou a través dun corredor. Desde a perspectiva da mensaxería, cada servizo ten varias fases de vida:

Inicialización do servizo.
Nesta fase, confírmase e lánzase o proceso de execución do servizo e as súas dependencias.
Creación dun punto de intercambio.
O servizo pode usar un punto de intercambio estático especificado na configuración do nodo ou crear puntos de intercambio de forma dinámica.
Rexistro do servizo.
Para que o servizo poida atender as solicitudes, debe estar rexistrado no punto de intercambio.
Funcionamento normal.
O servizo produce un traballo útil.
Apagado.
Existen 2 tipos de parada posibles: normal e de emerxencia. Durante o funcionamento normal, o servizo desconéctase do punto de intercambio e detense. En situacións de emerxencia, a mensaxería executa un dos scripts de failover.

Parece bastante complicado, pero o código non dá tanto medo. Exemplos de código con comentarios daranse na análise dos modelos un pouco máis adiante.

Intercambios

O punto de intercambio é un proceso de mensaxería que implementa a lóxica de interacción cos compoñentes dentro do modelo de mensaxería. En todos os exemplos que se presentan a continuación, os compoñentes interactúan a través de puntos de intercambio, cuxa combinación forma a mensaxería.

Patróns de intercambio de mensaxes (MEP)

A nivel mundial, os patróns de intercambio pódense dividir en dous sentidos e unidireccionais. Os primeiros implican unha resposta a unha mensaxe entrante, os segundos non. Un exemplo clásico dun patrón bidireccional na arquitectura cliente-servidor é o patrón Solicitude-resposta. Vexamos o modelo e as súas modificacións.

Solicitude-resposta ou RPC

RPC úsase cando necesitamos recibir unha resposta doutro proceso. Este proceso pode estar executado no mesmo nodo ou situado nun continente diferente. A continuación móstrase un diagrama da interacción entre o cliente e o servidor a través da mensaxería.

Dado que a mensaxería é completamente asíncrona, para o cliente o intercambio divídese en 2 fases:

Enviando solicitude
```
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
cambio ‒ nome único do punto de intercambio
Etiqueta de coincidencia de respostas ‒ etiqueta local para procesar a resposta. Por exemplo, no caso de enviar varias solicitudes idénticas pertencentes a diferentes usuarios.
Solicitude de definición - Órgano de solicitude
Proceso Handler ‒ PID do manipulador. Este proceso recibirá unha resposta do servidor.

Procesando a resposta

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

Carga útil de resposta - resposta do servidor.

Para o servidor, o proceso tamén consta de 2 fases:

Iniciando o punto de intercambio
Tramitación de solicitudes recibidas

Ilustremos este modelo con código. Digamos que necesitamos implementar un servizo sinxelo que proporcione un único método de hora exacta.

Código do servidor

Imos definir a API do servizo 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{}
}).

Imos definir o controlador de servizo 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.

Código de cliente

Para enviar unha solicitude ao servizo, podes chamar á API de solicitude de mensaxería en calquera lugar do cliente:

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

Nun sistema distribuído, a configuración dos compoñentes pode ser moi diferente e, no momento da solicitude, é posible que a mensaxería aínda non se inicie ou o controlador do servizo non estea preparado para atender a solicitude. Polo tanto, necesitamos comprobar a resposta da mensaxería e xestionar o caso de fallo.
Despois do envío exitoso, o cliente recibirá unha resposta ou erro do servizo.
Imos xestionar os dous 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};

Solicitude de resposta fragmentada

É mellor evitar enviar mensaxes enormes. Diso depende a capacidade de resposta e o funcionamento estable de todo o sistema. Se a resposta a unha consulta ocupa moita memoria, é obrigatorio dividila en partes.

Permíteme darche un par de exemplos deste tipo:

Os compoñentes intercambian datos binarios, como ficheiros. Dividir a resposta en pequenas partes axúdache a traballar de forma eficiente con ficheiros de calquera tamaño e evitar desbordamentos de memoria.
Listados. Por exemplo, necesitamos seleccionar todos os rexistros dunha táboa enorme na base de datos e transferilos a outro compoñente.

A estas respostas chamo locomotora. En calquera caso, 1024 mensaxes de 1 MB son mellores que unha única mensaxe de 1 GB.

No clúster Erlang, obtemos un beneficio adicional: reducir a carga no punto de intercambio e na rede, xa que as respostas envíanse inmediatamente ao destinatario, evitando o punto de intercambio.

Resposta con Solicitude

Esta é unha modificación bastante rara do patrón RPC para construír sistemas de diálogo.

Os sistemas impulsados por eventos entréganllas aos consumidores tan pronto como os datos estean listos. Así, os sistemas son máis propensos a un modelo push que a un modelo pull ou sondaxe. Esta función permítelle evitar desperdiciar recursos solicitando e esperando datos constantemente.
A figura mostra o proceso de distribución dunha mensaxe aos consumidores subscritos a un tema específico.

Exemplos clásicos de uso deste patrón son a distribución do estado: o mundo do xogo nos xogos de ordenador, os datos do mercado sobre os intercambios, a información útil nas fontes de datos.

Vexamos o código do abonado:

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.

A fonte pode chamar á función para publicar unha mensaxe en calquera lugar conveniente:

messaging:publish_message(Exchange, Key, Message).

cambio - nome do punto de intercambio,
Clave - clave de enrutamento
mensaxe - carga útil

Ao expandir pub-sub, podes obter un patrón conveniente para o rexistro. O conxunto de fontes e consumidores pode ser completamente diferente. A figura mostra un caso cun consumidor e varias fontes.

Patrón de distribución de tarefas

Case todos os proxectos implican tarefas de procesamento diferido, como a xeración de informes, a entrega de notificacións e a recuperación de datos de sistemas de terceiros. O rendemento do sistema que realiza estas tarefas pódese escalar facilmente engadindo controladores. Só nos queda formar un clúster de procesadores e repartir uniformemente as tarefas entre eles.

Vexamos as situacións que xorden usando o exemplo de 3 manipuladores. Mesmo na fase de distribución de tarefas, xorde a cuestión da equidade de distribución e o desbordamento dos manipuladores. A distribución en todo o mundo será a responsable da equidade, e para evitar unha situación de desbordamento dos manipuladores, introduciremos unha restrición prefetch_limit. En condicións transitorias prefetch_limit impedirá que un xestor reciba todas as tarefas.

A mensaxería xestiona as colas e a prioridade de procesamento. Os procesadores reciben tarefas a medida que chegan. A tarefa pode completarse con éxito ou fallar:

messaging:ack(Tack) - chamado se a mensaxe se procesou correctamente
messaging:nack(Tack) - Chamado en todas as situacións de emerxencia. Unha vez que se devolve a tarefa, a mensaxería pasaráaa a outro controlador.

Supoñamos que se produciu un fallo complexo ao procesar tres tarefas: o procesador 1, despois de recibir a tarefa, fallou sen ter tempo para informar nada ao punto de intercambio. Neste caso, o punto de intercambio transferirá a tarefa a outro controlador despois de que caduque o tempo de espera da confirmación. Por algún motivo, o xestor 3 abandonou a tarefa e enviou nack; como resultado, a tarefa tamén foi transferida a outro xestor que a completou con éxito.

Resumo preliminar

Cubrimos os bloques básicos dos sistemas distribuídos e adquirimos unha comprensión básica do seu uso en Erlang/Elixir.

Ao combinar patróns básicos, pode construír paradigmas complexos para resolver problemas emerxentes.

Na parte final da serie, analizaremos cuestións xerais de organización de servizos, enrutamento e equilibrio, e tamén falaremos do lado práctico da escalabilidade e da tolerancia a fallos dos sistemas.

Fin da segunda parte.

foto Marius Christensen
Ilustracións elaboradas mediante websequencediagrams.com

Fonte: www.habr.com

Bloques de construción de aplicacións distribuídas. Primeira aproximación