Esta é uma tradução do post A Guide to Event-Driven Architecture in Elixir escrito por Sapan Diwakar.

Também acrescentei algumas imagens para exemplificar melhor a explicação. Na dúvida sobre a tradução visite o post original. Vamos ao post.

Neste post, exploraremos como a arquitetura orientada a eventos (event-driven architecture) pode tornar sua aplicação mais responsiva aos usuários e desacoplar seus módulos para uma melhor experiência do desenvolvedor.

Também analisaremos vários métodos de implementação de arquitetura orientada a eventos com Elixir. A linguagem Elixir é particularmente boa para isso devido às APIs de passagem de mensagens avançadas e concisas que ela oferece e ao excelente suporte da BEAM para a concorrência.

Mas primeiro: o que é exatamente uma arquitetura orientada a eventos?

• Arquitetura orientada a eventos: Uma Introdução
• Construindo blocos de uma arquitetura orientada a eventos
• Benefícios da arquitetura orientada a eventos
• Abordagens
  • Arquitetura síncrona orientada a eventos em Elixir
  • Usando o GenServer na arquitetura orientada a eventos em Elixir
  • Implementação orientada a eventos usando GenStage em Elixir
• Conclusão: Arquitetura orientada a eventos em Elixir - Indo além do GenStage

Arquitetura orientada a eventos: Uma Introdução

A arquitetura orientada a eventos é uma arquitetura onde os eventos controlam o comportamento e o fluxo de sua aplicação. Os principais componentes da arquitetura são os produtores de eventos (event producers), o barramento de eventos (event bus) e os consumidores de eventos (event consumers).

Um evento pode ser qualquer coisa que represente uma mudança de estado no sistema. Por exemplo, em uma aplicação de comércio eletrônico ‘e-commerce’, a compra de um produto pelo usuário poderia produzir um evento de venda ao qual o consumidor poderia então iniciar um processo de atualização de estoque.

Uma arquitetura baseada em eventos permite que as aplicações atuem em eventos à medida que eles ocorrem. Diferentes partes de uma aplicação funcionam e se desenvolvem de forma relativamente independente em um projeto bem elaborado, baseado em eventos. As organizações podem designar equipes separadas para partes focadas da aplicação e racionalizar o fluxo de trabalho. Isto também cria uma fronteira clara entre diferentes partes de uma aplicação, auxiliando em futuros exercícios de escalabilidade.

A arquitetura orientada por eventos ganhou popularidade principalmente com produtos baseados em microserviços, mas também pode ser usada para um monólito.

As coisas são sempre mais claras com um exemplo, então vamos olhar para um.

Construindo blocos de uma arquitetura orientada a eventos

Vamos discutir cada bloco de construção em detalhes, usando uma aplicação de ‘e-commerce’ como exemplo.

Imagine que um usuário faz uma nova compra em um ‘website’. O novo pedido é um evento gerado pela parte que controla o pedido: o event producer.

O evento pode ser empurrado para um event bus. O event bus pode ser qualquer coisa, como, por exemplo:

• Uma tabela no banco de dados.
• Uma fila de eventos em memória dentro da aplicação.
• Uma ferramenta externa como RabbitMQ ou Apache Kafka.

Os event producers interessados neste tipo de evento podem se inscrever no event bus. O evento é entregue-lhes, e eles fazem algum processamento em cima dele.

Por exemplo, um sistema de gerenciamento de estoque assinaria (subscribe) o novo evento de pedido e atualizaria o estoque do produto. Outro sistema também poderia escolher o mesmo evento no pedido - por exemplo, um serviço de atendimento poderia processar esse evento e criar uma rota de entrega para o produto.

Benefícios da arquitetura orientada a eventos

Há várias vantagens em utilizar uma arquitetura orientada a eventos ao invés de uma que processa tudo sequencialmente.

Uma arquitetura orientada a eventos nos permite construir várias partes independentes de uma aplicação para trabalhar com o mesmo evento e realizar diferentes tarefas focadas. Isto pode ser vantajoso por um par de razões:

• Uma equipe precisa se concentrar apenas em uma parte.
• O código de aplicação para partes pequenas pode ser simples.

Outra vantagem do conceito é que ele nos permite fornecer uma ‘interface’ rápida ao usuário. Como exemplo da aplicação acima, um usuário precisa fazer um pedido. A aplicação pode continuar processando todas as outras tarefas não-interativas, como atualizar seu inventário e interagir com a aplicação de entrega, sem a atenção do usuário.

Isto também torna muito fácil acrescentar novas etapas de processamento no pipeline do evento. Por exemplo, digamos que precisamos de uma tarefa adicional a ser realizada a partir de um evento. Só precisamos adicionar um novo consumidor para lidar com o evento, sem tocar em nenhuma outra parte da aplicação.

Finalmente, é muito mais fácil escalar cada módulo individual se ele for desacoplado dos outros, do que escalar uma aplicação inteira em conjunto. Isto é ainda mais benéfico ao ter uma parte que consome muito mais recursos do que suas contrapartes no pipeline de processamento de eventos.

Mas a arquitetura orientada a eventos não vem sem desvantagens quando não é bem pensada. Se aplicada a problemas muito simples, ela pode levar a fluxos de trabalho complexos que são lentos e difíceis de depurar.

Vamos explorar algumas maneiras simples de implementar uma arquitetura orientada a eventos com Elixir, sem a necessidade de escrever partes complexas de código.

Arquitetura síncrona orientada a eventos em Elixir

A maneira mais simples (e mais ineficiente) de executar o fluxo acima seria fazer tudo de forma síncrona no pedido do usuário.

Portanto, se você tiver um módulo de Pedidos (Orders) que processe o pedido de um usuário, a implementação síncrona poderia se parecer com isto:

defmodule Orders do
  def create_order(attrs) do
    # salva o pedido
    {:ok, order} = save_order(attrs)
    # atualiza o estoque
    {:ok, _inventory} = update_inventory(order)
    # cria a entrega
    {:ok, _delivery} = create_delivery(order)
    # retorna o pedido
    {:ok, order}
  end
end

Podemos melhorar isto para que os consumidores de novos eventos possam escalar mais facilmente:

defmodule Orders do
  @event_consumers [
    {Inventory, :handle_event},
    {Delivery, :handle_event},
  ]

  def create_order(attrs) do
    {:ok, order} = save_order(attrs)

    event = %Orders.Event{type: :new_order, payload: order}
    @event_consumers
    |> Enum.each(fn {module, func} ->
      apply(module, func, [event])
    end)

    {:ok, order}
  end
end

Com esta implementação, tudo o que precisamos fazer para adicionar novos consumidores é adicionar a especificação na lista @event_consumers, e esses consumidores podem trabalhar independentemente.

Embora a abordagem síncrona funcione bem para um pequeno número de consumidores, ela tem uma desvantagem. A criação de um pedido pode levar muito tempo porque será necessário esperar a atualização do estoque e a criação da entrega.

Estas são tarefas internas que podem ser realizadas sem interação do usuário e movidas da cadeia síncrona.

Vamos ver como podemos racionalizar ainda mais a arquitetura orientada a eventos usando o GenServer.

Usando o GenServer na arquitetura orientada a eventos em Elixir

Para esta implementação, executaremos um processo separado para cada consumidor. Estes se inscreverão num evento de um produtor e executarão suas tarefas simultaneamente.

Para o event bus, podemos utilizar o Phoenix.PubSub. Note que é possível para aplicativos que não usam Phoenix usar diretamente o Registro como um PubSub.

Primeiro, vamos olhar para o produtor.

defmodule Orders do
  def create_order(attrs) do
    {:ok, order} = save_order(attrs)
    event = %Orders.Event{type: :new_order, payload: order}
    Phoenix.PubSub.broadcast(:my_app, "new_order", event)
    {:ok, order}
  end
end

Em um novo pedido (order), criamos a estrutura do evento e usamos o Phoenix.PubSub.broadcast/3 para transmitir esse evento no barramento (bus). Como você pode ver, é muito mais simples do que a implementação anterior, onde o módulo de Orders processava tarefas do outro módulo em série.

Os consumidores podem então assinar (subscribe) o tópico new_order e implementar o handle_info/2 para serem notificados toda vez que um novo evento é publicado pelo produtor.

defmodule Inventory do
  use GenServer

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)

  def init(_opts) do
    Phoenix.PubSub.subscribe(:my_app, "new_order")
  end

  def handle_info(%Orders.Event{type: :new_order, payload: order}, state) do
    state = consume(state, order.product)
    {:noreply, state}
  end
end

O módulo de Delivery será muito semelhante ao acima mencionado, por isso estou pulando aqui.

Como você pode ver, isto é muito melhor do que as implementações anteriores. Os módulos de Inventory e Delivery podem se inscrever independentemente para o tópico new_order. O módulo Orders transmite (broadcasts) para este tópico sobre novos pedidos e eventos são entregues para os processos inscritos.

Você pode até distribuir isto entre vários nós e Phoenix.PubSub (com um PG, Redis ou outro adaptador), espalhando os eventos para todos os nós.

Ótimo, certo? Na verdade, não. Há vários problemas com esta abordagem:

• O PubSub fornece uma transmissão em tempo real sem fila de mensagens, portanto, se um dos processos do assinante estiver em suspenso, ele pode perder as transmissões.

• Se o assinante fizer algum trabalho pesado, ele pode não conseguir acompanhar as mensagens recebidas, resultando em timeouts e, conseqüentemente, em um colapso da árvore de processos.

• Se o assinante encontrar um erro no processamento de uma mensagem, ela é considerada consumida e não será novamente processada (retried) mais tarde.

Portanto, esta abordagem é uma má abordagem a ser seguida para nosso caso de uso atual.

Entretanto, a abordagem ainda tem seus casos de uso. Ela pode ser usada para tarefas que não são críticas, ou que podem ser corrigidas com a próxima mensagem: por exemplo, se uma tarefa computar as sugestões para as próximas compras de um usuário com base em sua última compra. Embora isto também precise ser acionado em um novo pedido, não é exatamente crítico (para um site tradicional de e-commerce) e pode recalcular as sugestões para um usuário em sua próxima compra.

Implementação orientada a eventos usando GenStage em Elixir

Na seção anterior, vimos uma grande implementação de nosso sistema orientado a eventos utilizando o GenServer. Mas ele não veio sem suas limitações. Vamos ver como o GenStage se comporta.

O GenStage faz uma distinção clara entre producers, consumers e producer_consumers, e cada processo tem que escolher um quando começa (em seu init/1). Em nosso caso, tanto o Inventory quanto o Delivery são consumers, e Orders é um producer.

É aqui que as coisas começam a ficar um pouco complicadas. O GenStage tem um conceito de demanda. Cada consumer pode emitir uma demanda de quantos eventos ele pode lidar. O producer precisa enviar esses eventos para o consumer. Vamos ver um producer básico em ação.

defmodule Orders do
  use GenStage

  def start_link(opts) do
    GenStage.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(_opts) do
    {:producer, :some_state_which_does_not_currently_mattere}
  end

  def create_order(pid, attrs) do
    GenStage.cast(pid, {:create_order, attrs})
  end

  def handle_cast({:create_order, attrs}, state) do
    {:ok, order} = save_order(attrs)
    {:noreply, [%Orders.Event{type: new_order, payload: order}], state}
  end

  def handle_demand(_demand, state), do: {:noreply, [], state}
end

A carne de nosso código está em handle_cast, onde guardamos o pedido (order) e devolvemos uma tupla como {:noreply, events, new_state}. Os novos eventos são armazenados em um buffer GenStage interno e enviados aos consumidores à medida que eles fazem novas demandas (ou imediatamente, se houver consumidores com demanda não atendida).

Vamos checar uma amostra da implementação do consumer:

defmodule Inventory do
  use GenStage

  def start_link(opts) do
    GenStage.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(_opts) do
    {:consumer, [], subscribe_to: [Orders]}
  end

  def handle_events(events, _from, state) do
    state = Enum.reduce(events, state, & handle_event(&1, &2))
    {:noreply, [], state}
  end

  def handle_event(%Orders.Event{type: :new_order, payload: order}, state) do
    new_state = update_inventory(order)
    new_state
  end
end

No consumer, primeiro avisa de que temos um subscribe_to (inscrição_para) dentro do init/1. Isto automaticamente assina/inscreve (subscribe) o Inventory para qualquer evento publicado por Orders. Favor verificar a documentação GenStage para opções adicionais disponíveis no init.

Aqui, a maior parte do trabalho acontece dentro do handle_events/3, que é automaticamente chamado pela GenStage assim que novos eventos se tornam disponíveis. Tratamos aqui do evento new_order, atualizando o inventory e retornando um novo estado.

Com esta simples implementação, obtemos vários benefícios que superam a implementação do GenServer:

• Armazenamento automático de eventos dentro do buffer interno do GenStage quando o producer tem novos eventos sem nenhum consumer disponível.

Mesmo que os consumers estejam ausentes quando alguns eventos são produzidos, nós ainda temos a garantia de recebê-los quando o consumer volta a ficar disponível.

Confira o guia da Genstage sobre a demanda de buffering para uma lógica de buffering avançada.

• Distribuição automática de trabalho em múltiplos consumidores.

Se você tem tarefas pesadas de consumo, você pode iniciar múltiplos processos de consumo. O DemandDispatcher padrão para GenStage distribuirá o trabalho uniformemente por todos os processos.

Veja o GenStage.Dispatcher para outras estratégias de despacho para distribuir eventos a todos os consumidores ou distribuição de partições aos consumidores com base em uma função hash.

Mas, como no GenServer ou na implementação síncrona, o uso do GenStage não vem sem seus problemas.

Se um consumer se chocar enquanto estiver processando um evento, o GenStage considerará o evento entregue e não o enviará novamente quando o consumer voltar a ficar disponível.

Para ter certeza de que você rastreia corretamente os acidentes, você pode usar um serviço de monitoramento como o AppSignal. O AppSignal é [fácil de instalar para sua aplicação Elixir] e ajuda a monitorar o desempenho, assim como rastrear erros. Aqui está um exemplo de um painel de controle de erros que o AppSignal fornece:

Você também pode configurar notificações de colisões via AppSignal.

No lado do aplicativo, você pode armazenar tais eventos em uma loja persistente uma vez que eles sejam entregues ao consumidor (consumer). Se o consumidor se chocar, então uma vez que se recupere, pode retornar aos eventos em cache.

Seja muito cauteloso em produzir muitos eventos sem consumidores suficientes, no entanto. Enquanto a GenStage oferece um buffer automático de eventos, este buffer tem um tamanho máximo (configurável) e um tamanho máximo prático limitado pela memória do servidor.

Se você não controla a freqüência com que os eventos são produzidos, considere o uso de um armazenamento de dados externo como Redis ou Postgres para fazer o buffer de eventos.

Conclusão: Arquitetura orientada a eventos em Elixir - Indo além do GenStage

Neste post, examinamos três abordagens para implementar um sistema orientado a eventos no Elixir: de forma síncrona, usando o GenServer, e finalmente, usando o GenStage. Examinamos algumas das vantagens e desvantagens de cada abordagem.

O exemplo simples do GenStage pode ser um ponto de partida para a implementação de complexos pipelines de processamento de dados orientados por eventos que abrangem vários nós. Sugiro que você leia a grande documentação do GenStage para obter mais informações.

Se você estiver procurando por uma abstração ainda maior, a Broadway é um bom ponto de partida. Ela é construída sobre o GenStage e oferece várias características adicionais, incluindo o consumo de dados de filas externas como Amazon SQS, Apache Kafka, e RabbitMQ.

Até a próxima vez, boa codificação!