🥇Blocos de construção de aplicativos distribuídos. Primeiro aproximar

Na parte anterior artigo Analisamos as bases teóricas da arquitetura reativa. É hora de falar sobre fluxos de dados, formas de implementação de sistemas reativos Erlang/Elixir e padrões de troca de mensagens neles:

Resposta de requisição
Resposta de requisição em pedaços
Resposta com requisição
Publicar-assinar
Publicação-assinatura invertida
Distribuição de tarefas

SOA, MSA e troca de mensagens

SOA e MSA são arquiteturas de sistemas que definem as regras para a construção de sistemas, enquanto a troca de mensagens fornece os primitivos para a sua implementação.

Não quero promover uma ou outra arquitetura para a construção de sistemas. Eu defendo a aplicação das práticas mais eficientes e úteis para um projeto e negócio específicos. Independentemente da paradigma que escolhermos, é melhor criar blocos de sistemas com a visão do modo Unix: componentes com baixa acoplamento, responsáveis por entidades individuais. Os métodos da API executam as ações mais simples possíveis com entidades.

Messaging ‒ como o nome sugere ‒ é um broker de mensagens. Seu objetivo principal é receber e enviar mensagens. Ele é responsável pelas interfaces de envio de informações, formação de canais lógicos de transmissão de informações dentro do sistema, roteamento e balanceamento, bem como no tratamento de falhas em nível de sistema.
O messaging em desenvolvimento não tenta competir com o rabbitmq ou substituí-lo. Suas principais características são:

Distribuição.
Pontos de troca podem ser criados em todos os nós do cluster, o mais próximo possível do código que os utiliza.
Simplicidade.
Foco na minimização de código padrão e facilidade de uso.
Melhor desempenho.
Não tentamos replicar a funcionalidade do rabbitmq, mas destacamos apenas a camada arquitetônica e de transporte, que se integra da maneira mais simples possível no OTP, minimizando os custos.
Flexibilidade.
Cada serviço pode incorporar vários padrões de troca.
Resiliência, incorporada ao design.
Escalabilidade.
O messaging cresce junto com a aplicação. À medida que a carga aumenta, os pontos de troca podem ser movidos para máquinas separadas.

Observação. Do ponto de vista da organização do código, meta-projetos são bem adequados para sistemas complexos em Erlang/Elixir. Todo o código do projeto está em um único repositório ‒ projeto sombrinha. Os microsserviços, por sua vez, são maximizados em isolamento e realizam operações simples, responsáveis por uma entidade separada. Com essa abordagem, é fácil manter a API de todo o sistema, realizar alterações de forma simples e escrever testes unitários e de integração de forma conveniente.

Os componentes do sistema interagem diretamente ou via um broker. Do ponto de vista de messaging, cada serviço possui várias fases de vida:

Inicialização do serviço.
Nesta fase, ocorre a configuração e o início do processo de execução do serviço e suas dependências.
Criação de um ponto de troca.
O serviço pode usar um ponto de troca estático, definido na configuração do nó, ou criar pontos de troca dinamicamente.
Registro do serviço.
Para que o serviço possa atender a solicitações, ele precisa ser registrado no ponto de troca.
Funcionamento normal.
O serviço realiza trabalho útil.
Finalização do trabalho.
Existem 2 tipos de finalização: normal e anômala. Na finalização normal, o serviço se desconecta do ponto de troca e é interrompido. Em casos anômalos, o messaging executa um dos cenários de tratamento de falhas.

Parece bastante complexo, mas no código não é tão assustador. Exemplos de código com comentários serão apresentados na análise dos templates um pouco mais adiante.

Exchanges

O ponto de troca é um processo de messaging que implementa a lógica de interação com os componentes dentro do template de troca de mensagens. Em todos os exemplos apresentados abaixo, os componentes interagem através de pontos de troca, cuja combinação forma o messaging.

Message exchange patterns (MEPs)

Globalmente, os templates de troca podem ser divididos em bidirecionais e unidirecionais. Os primeiros implicam uma resposta à mensagem recebida, enquanto os segundos não. Um exemplo clássico de template bidirecional na arquitetura cliente-servidor é o template Request-response. Vamos considerar este template e suas modificações.

Request–response ou RPC

RPC é utilizado quando precisamos obter uma resposta de outro processo. Esse processo pode estar em execução no mesmo nó ou em outro continente. Abaixo está o diagrama de interação entre o cliente e servidor via messaging.

Como o messaging é completamente assíncrono, a troca é dividida em 2 fases para o cliente:

Envio de solicitação
```
mensageria:pedido(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
Exchange ‒ nome único do ponto de troca
ResponseMatchingTag ‒ etiqueta local para o processamento da resposta. Por exemplo, ao enviar várias solicitações iguais pertencentes a diferentes usuários.
RequestDefinition ‒ corpo da solicitação
HandlerProcess ‒ PID do manipulador. Esta é a instância que receberá a resposta do servidor.

Processamento da resposta

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

ResponsePayload ‒ resposta do servidor.

Para o servidor, o processo consiste em 2 fases:

Inicialização do ponto de troca
Processamento das solicitações recebidas

Vamos ilustrar este modelo com código. Suponha que precisamos implementar um serviço simples que fornece um único método para obter a hora exata.

Código do servidor

Vamos extrair a definição da API do serviço para api.hrl:

%% =====================================================
%%  entidades
%% =====================================================
-gravar(time, {
  unixtime :: inteiro_nao_negativo(),
  datetime :: binário()
}).

-gravar(time_error, {
  code :: inteiro_nao_negativo(),
  erro :: termo()
}).

%% =====================================================
%%  métodos
%% =====================================================
-gravar(time_req, {
  opts :: termo()
}).
-gravar(time_resp, {
  resultado :: #time{} | #time_error{}
}).

Definiremos o controlador do serviço em time_controller.erl

%% O exemplo mostra apenas o código significativo. Inserindo-o no modelo gen_server podemos obter um serviço funcional.

%% inicialização do gen_server
init(Args) ->
  %% conexão ao ponto de troca
  mensageria:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% processamento do evento de perda de conexão com o ponto de troca. Este mesmo evento chega se o ponto de troca ainda não tiver sido iniciado.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% processamento da API
handle_info(#time_req{opts = _Opts}, State) ->
  mensageria:response_once(Client, #time_resp{
resultado = #time{ unixtime = time_utils:unixtime(now()), datetime = time_utils:iso8601_fmt(now())}
  });
  {noreply, State};

%% finalização do gen_server
terminate(_Reason, _State) ->
  mensageria:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Código do cliente

Para enviar uma solicitação ao serviço, em qualquer lugar do cliente, você pode chamar a API de pedido da mensageria:

caso mensageria:request(?EXCHANGE, tag, #time_req{opts = #{}} , self()) do
    ok -> ok;
    _ -> %% lógica de repetição ou falha
fim

Em um sistema distribuído, a configuração dos componentes pode ser variada e, no momento da solicitação, a mensageria pode ainda não ter iniciado ou o controlador do serviço pode não estar pronto para atender à solicitação. Portanto, precisamos verificar a resposta da mensageria e tratar o caso de falha.
Após o envio bem-sucedido, o cliente receberá uma resposta ou um erro do serviço.
Vamos lidar com ambos os casos em 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};

Resposta de requisição em pedaços

É melhor evitar o envio de mensagens grandes. Isso afeta a responsividade e a estabilidade de todo o sistema. Se a resposta a uma solicitação ocupa muita memória, a fragmentação em partes é obrigatória.

Vou dar alguns exemplos desses casos:

Os componentes trocam dados binários, como arquivos. Dividir a resposta em partes menores ajuda a trabalhar eficientemente com arquivos de qualquer tamanho e evita estouros de memória.
Listagens. Por exemplo, precisamos selecionar todos os registros de uma enorme tabela no banco e passar para outro componente.

Eu chamo essas respostas de locomotivas. De qualquer forma, 1024 mensagens de 1 MB são melhores do que uma única mensagem de 1 GB.

Em um cluster Erlang, obtemos um ganho adicional ‒ diminuição da carga no ponto de troca e na rede, já que as respostas são enviadas diretamente ao destinatário, ignorando o ponto de troca.

Resposta com requisição

Esta é uma modificação bastante rara do padrão RPC para construir sistemas de diálogo.

Sistemas orientados a eventos entregam os dados aos consumidores assim que estão prontos. Assim, os sistemas tendem mais para um modelo de push do que para pull ou poll. Essa característica evita desperdiçar recursos pedindo e esperando dados constantemente.
A figura mostra o processo de disseminação da mensagem para os consumidores que assinaram um determinado tópico.

Exemplos clássicos do uso desse padrão incluem a disseminação de estado: do mundo do jogo em jogos de computador, dados de mercado em bolsas de valores, informações úteis em feeds de dados.

Vamos analisar o código do assinante:

init(_Args) ->
  %% assinamos o ponto de troca, chave = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% se o ponto de troca não estiver disponível, tentamos reconectar
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% processamos as mensagens recebidas
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% ao parar o consumidor - desconectamo-nos do ponto de troca
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

A fonte pode chamar a função de publicação de mensagem em qualquer lugar conveniente:

messaging:publish_message(Exchange, Key, Message).

Exchange ‒ nome do ponto de troca,
Chave ‒ chave de roteamento
Mensagem ‒ carga útil

Ao implementar pub-sub, obtém-se um padrão conveniente para registro em log. O conjunto de fontes e consumidores pode ser completamente diferente. A figura apresenta um caso com um consumidor e múltiplas fontes.

Padrão de distribuição de tarefas

Quase em todo projeto, surgem tarefas de processamento atrasado, como geração de relatórios, envio de notificações e obtenção de dados de sistemas externos. A capacidade do sistema que executa essas tarefas é facilmente escalável por meio da adição de processadores. Tudo o que nos resta é formar um cluster de processadores e distribuir as tarefas uniformemente entre eles.

Consideremos as situações que surgem no exemplo de 3 processadores. Mesmo na fase de distribuição das tarefas, surge a questão da justiça na distribuição e do esgotamento dos processadores. A justiça será garantida pela distribuição round-robin e, para evitar a situação de sobrecarga dos processadores, introduziremos um limite prefetch_limit. Nos modos de transição, prefetch_limit não permitirá que um processador receba todas as tarefas.

Messaging gerencia filas e prioridades de processamento. Os processadores recebem tarefas à medida que elas chegam. A execução de uma tarefa pode ser concluída com sucesso ou falhar:

messaging:ack(Tack) ‒ chamado em caso de processamento bem-sucedido da mensagem
messaging:nack(Tack) ‒ chamado em todas as situações não conforme. Após retornar a tarefa, o messaging a transfere para outro processador.

Suponha que, ao processar três tarefas, ocorreu uma falha complexa: o processador 1, após receber a tarefa, falhou, sem ter conseguido comunicar nada ao ponto de troca. Nesse caso, após o término do ack timeout, o ponto de troca transferirá a tarefa para outro processador. O processador 3, por algum motivo, rejeitou a tarefa e enviou nack; assim, a tarefa também foi transferida para outro processador que a executou com sucesso.

Resumo preliminar

Discutimos os principais blocos de construção de sistemas distribuídos e obtivemos uma compreensão básica de sua aplicação em Erlang/Elixir.

Combinando padrões básicos, é possível construir paradigmas complexos para resolver as tarefas que surgem.

Na parte final do ciclo, abordaremos questões gerais sobre a organização de serviços, roteamento e balanceamento, além de discutir o aspecto prático da escalabilidade e resiliência dos sistemas.

Fim da segunda parte.

Foto Marius Christensen
Ilustrações preparadas com o auxílio de websequencediagrams.com

Fonte: habr.com