Blocos de construção de aplicativos distribuídos. Primeira abordagem

No passado статье Examinamos os fundamentos teóricos da arquitetura reativa. É hora de falar sobre fluxos de dados, maneiras de implementar sistemas reativos Erlang/Elixir e padrões de mensagens neles:

• Solicitação-resposta
• Resposta fragmentada de solicitação
• Resposta com solicitação
• Publicar-assinar
• Publicação-assinatura invertida
• Distribuição de tarefas

SOA, MSA e mensagens

SOA, MSA são arquiteturas de sistema que definem as regras para a construção de sistemas, enquanto as mensagens fornecem primitivas para sua implementação.

Não quero promover esta ou aquela arquitetura de sistema. Defendo o uso das práticas mais eficazes e úteis para um projeto e negócio específico. Qualquer que seja o paradigma escolhido, é melhor criar blocos de sistema de olho no estilo Unix: componentes com conectividade mínima, responsáveis ​​por entidades individuais. Os métodos API executam as ações mais simples possíveis com entidades.

Mensagens é, como o nome sugere, um corretor de mensagens. Sua principal finalidade é receber e enviar mensagens. É responsável pelas interfaces de envio de informações, pela formação de canais lógicos de transmissão de informações dentro do sistema, pelo roteamento e balanceamento, bem como pelo tratamento de falhas no nível do sistema.
A mensagem que estamos desenvolvendo não tenta competir ou substituir o RabbitMQ. Suas principais características:

• Distribuição.
  Os 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.
  Concentre-se em minimizar o código padrão e na facilidade de uso.
• A melhor performance.
  Não pretendemos repetir a funcionalidade do RabbitMQ, mas destacar apenas a camada arquitetônica e de transporte, que encaixamos no OTP da forma mais simples possível, minimizando custos.
• Flexibilidade.
  Cada serviço pode combinar vários modelos de troca.
• Resiliência desde o design.
• Escalabilidade.
  As mensagens crescem com o aplicativo. À medida que a carga aumenta, você pode mover os pontos de troca para máquinas individuais.

Observação Em termos de organização de código, metaprojetos são adequados para sistemas Erlang/Elixir complexos. Todo o código do projeto está localizado em um repositório - um projeto guarda-chuva. Ao mesmo tempo, os microsserviços são isolados ao máximo e executam operações simples que são responsáveis ​​​​por uma entidade separada. Com esta abordagem, é fácil manter a API de todo o sistema, é fácil fazer alterações, é conveniente escrever testes unitários e de integração.

Os componentes do sistema interagem diretamente ou através de um corretor. Do ponto de vista das mensagens, cada serviço tem várias fases de vida:

• Inicialização do serviço.
  Nesta fase, o processo que executa o serviço e suas dependências são configurados e lançados.
• Criando um ponto de troca.
  O serviço pode usar um ponto de troca estático especificado na configuração do nó ou criar pontos de troca dinamicamente.
• Cadastro de serviço.
  Para que o serviço atenda às solicitações, ele deve ser cadastrado no ponto de troca.
• Funcionamento normal.
  O serviço produz trabalho útil.
• Desligar.
  Existem 2 tipos de desligamento possíveis: normal e de emergência. Durante a operação normal, o serviço é desconectado do ponto de troca e para. Em situações de emergência, o sistema de mensagens executa um dos scripts de failover.

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

Trocas

O ponto de troca é um processo de mensagens que implementa a lógica de interação com componentes dentro do modelo de mensagens. Em todos os exemplos apresentados a seguir, os componentes interagem por meio de pontos de troca, cuja combinação forma mensagens.

Padrões de troca de mensagens (MEPs)

Globalmente, os padrões de troca podem ser divididos em bidirecionais e unidirecionais. Os primeiros implicam uma resposta a uma mensagem recebida, os últimos não. Um exemplo clássico de padrão bidirecional na arquitetura cliente-servidor é o padrão Solicitação-resposta. Vejamos o modelo e suas modificações.

Solicitação-resposta ou RPC

O RPC é usado quando precisamos receber uma resposta de outro processo. Este processo pode estar em execução no mesmo nó ou localizado em um continente diferente. Abaixo está um diagrama da interação entre cliente e servidor por meio de mensagens.

Como o envio de mensagens é totalmente assíncrono, para o cliente a troca é dividida em 2 fases:

1. Enviando uma solicitação

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

   Exchange ‒ nome exclusivo do ponto de troca
   ResponseMatchingTag ‒ rótulo local para processar a resposta. Por exemplo, no caso de envio de vários pedidos idênticos pertencentes a utilizadores diferentes.
   Definição de solicitação - solicitar corpo
   Processo Manipulador ‒ PID do manipulador. Este processo receberá uma resposta do servidor.

2. Processando a resposta

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

   ResponsePayload - resposta do servidor.

Para o servidor, o processo também consiste em 2 fases:

1. Inicializando o ponto de troca
2. Processamento de solicitações recebidas

Vamos ilustrar este modelo com código. Digamos que precisamos implementar um serviço simples que forneça um único método de hora exata.

Código do servidor

Vamos definir a API do serviço em 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{}
}).

Vamos definir o controlador de serviço em 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 do cliente

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

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

Em um sistema distribuído, a configuração dos componentes pode ser muito diferente e no momento da solicitação, o sistema de mensagens ainda não pode ser iniciado ou o controlador de serviço não estará pronto para atender a solicitação. Portanto, precisamos verificar a resposta da mensagem e tratar o caso de falha.
Após o envio bem-sucedido, o cliente receberá uma resposta ou 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 fragmentada de solicitação

É melhor evitar enviar mensagens enormes. A capacidade de resposta e a operação estável de todo o sistema dependem disso. Se a resposta a uma consulta ocupar muita memória, será obrigatório dividi-la em partes.

Deixe-me dar alguns exemplos de tais casos:

• Os componentes trocam dados binários, como arquivos. Dividir a resposta em pequenas partes ajuda você a trabalhar de forma eficiente 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 de dados e transferi-los para outro componente.

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

No cluster Erlang, obtemos um benefício adicional - redução da carga no ponto de troca e na rede, pois as respostas são enviadas imediatamente ao destinatário, contornando o ponto de troca.

Resposta com solicitação

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

Publicar-assinar (árvore de distribuição de dados)

Os sistemas orientados a eventos os entregam aos consumidores assim que os dados estão prontos. Assim, os sistemas são mais propensos a um modelo push do que a um modelo pull ou poll. Esse recurso permite evitar o desperdício de recursos solicitando e aguardando dados constantemente.
A figura mostra o processo de distribuição de uma mensagem aos consumidores inscritos em um tema específico.

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

Vejamos o código do assinante:

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 a função para publicar uma mensagem em qualquer local conveniente:

messaging:publish_message(Exchange, Key, Message).

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

Publicação-assinatura invertida

Ao expandir o pub-sub, você pode obter um padrão conveniente para registro. O conjunto de fontes e consumidores pode ser completamente diferente. A figura mostra um caso com um consumidor e múltiplas fontes.

Padrão de distribuição de tarefas

Quase todos os projetos envolvem tarefas de processamento diferido, como geração de relatórios, entrega de notificações e recuperação de dados de sistemas de terceiros. A taxa de transferência do sistema que executa essas tarefas pode ser facilmente dimensionada adicionando manipuladores. Tudo o que nos resta é formar um cluster de processadores e distribuir uniformemente as tarefas entre eles.

Vejamos as situações que surgem usando o exemplo de 3 manipuladores. Mesmo na fase de distribuição de tarefas, surge a questão da justiça na distribuição e do excesso de manipuladores. A distribuição round-robin será responsável pela justiça e, para evitar uma situação de excesso de manipuladores, introduziremos uma restrição pré-busca_limit. Em condições transitórias pré-busca_limit impedirá que um manipulador receba todas as tarefas.

O sistema de mensagens gerencia filas e prioridade de processamento. Os processadores recebem as tarefas conforme elas chegam. A tarefa pode ser concluída com êxito ou falhar:

• messaging:ack(Tack) - chamado se a mensagem for processada com sucesso
• messaging:nack(Tack) - chamado em todas as situações de emergência. Assim que a tarefa for retornada, as mensagens a passarão para outro manipulador.

Suponha que ocorreu uma falha complexa durante o processamento de três tarefas: o processador 1, após receber a tarefa, travou sem ter tempo de reportar nada ao ponto de troca. Nesse caso, o ponto de troca transferirá a tarefa para outro manipulador após o tempo limite de confirmação expirar. Por alguma razão, o manipulador 3 abandonou a tarefa e enviou nack; como resultado, a tarefa também foi transferida para outro manipulador que a completou com sucesso.

Resultado preliminar

Cobrimos os blocos básicos de construção de sistemas distribuídos e obtivemos uma compreensão básica de seu uso em Erlang/Elixir.

Ao combinar padrões básicos, você pode construir paradigmas complexos para resolver problemas emergentes.

Na parte final da série, consideraremos questões gerais de organização de serviços, roteamento e balanceamento, e também falaremos sobre o lado prático da escalabilidade e tolerância a falhas dos sistemas.

Fim da segunda parte.

foto Marius Christensen
Ilustrações preparadas usando websequencediagrams.com

Fonte: habr.com