CQRS e Event Sourcing com Axon Framework e Spring Boot

Olá pessoal, hoje vou dar um overview de como o Axon Framework pode te ajudar a implementar de forma simples uma aplicação orientada a eventos seguindo o padrão CQRS(Command Query Responsibility Segregation). Vamos implementar uma pequena aplicação que irá simular um controlador de contas bancárias. A ideia é conseguir passar de forma clara como os eventos em uma escala temporal contínua conseguirão reproduzir um resultado coeso e auditável.

CQRS

A essência do CQRS, que significa Segregação por Responsabilidade de Consulta e Comando, é utilizar modelos diferentes para atualizar e ler informações.

A maioria dos sistemas hoje em dia são tratados como CRUDs, ou seja, utilizamos um único modelo para realizar operações de escrita, edição, leitura e quando não precisamos mais deles, a exclusão. Geralmente tentamos manter o modelo retornado o mais fiel possível ao usado pelo armazenamento.

Conforme o tempo as necessidades do sistema vão ficando mais complexas e começamos a nos afastar desse modelo. Não é muito difícil começarmos a querer retornar dados agrupados ou no caso de uma atualização receber dados que são usados apenas para validações. Isso faz com que várias representações da mesma informação sejam criadas deixando o modelo cada vez mais carregado e complexo.

Exemplo de aplicação "single model"

O CQRS propõe que separemos a aplicação em modelos diferentes para atualização e exibição, que seguindo o padrão CQRS são os Comandos e Consultas. Quando digo separados quero dizer com objetos diferentes, em processos separados, as vezes até em hardware separado.

O exemplo que vamos abordar nessa Story é um cadastro de contas bancárias. No momento que o usuário solicita uma mudança isso é roteado para o modelo de comando que realizará as alterações necessárias e informará o modelo de consulta para que as novas informações sejam exibidas para o usuário.

Isso pode ser feito de várias formas. Os modelos na memória podem compartilhar um mesmo banco de dados ou podem ter bases separadas, mas para isso precisaríamos de um método diferente para manter a comunicação entre os dois modelos. E esse problema dá o gancho para o nosso próximo tópico.

Event Sourcing

Quando trabalhamos com um modelo tradicional conseguimos consultar em nosso banco de dados somente o último estado de um objeto e em muitos casos isso basta, porém em alguns momentos somente o estado atual não basta, precisamos saber também como foi o caminho para chegar-mos nesse resultado.

O Event Sourcing garante que todas as alterações em uma aplicação sejam armazenadas em uma sequencia de eventos. Assim podemos montar projeções do estado atual da nossa aplicação assim como consultar os eventos para montar projeções do passado ou para auditorias.

Trazendo para o nosso exemplo, imagine que uma conta bancária seja criada, depois editada varias vezes e por fim fechada/removida. Se estivermos olhando somente para o estado final da conta conseguimos verificar que a mesma está com o status CLOSED e um saldo de R$ 15,00 por exemplo. Porém isso não nos dá informação suficiente, tal como o porque a conta foi fechada.

Quando trabalhamos com Event Sourcing temos em mãos todo o histórico de eventos que constituem o estado atual dessa conta, e conseguimos verificar se a mesma não tinha movimentação, e por isso foi fechada. Poderíamos tratar inconsistências no estado atual de uma conta reportada pelo usuário simplesmente olhando para o passado e montando seu estado novamente.

Há muitas vantagens em utilizar uma arquitetura orientada a eventos, mas antes de decidir utiliza-la, você deve verificar se todos esses benefícios fazem sentido para a sua aplicação. Uma arquitetura orientada e eventos pode aumentar muito a complexidade da aplicação, tanto de implementação quanto de manutenção.

Axon Framework

O Axon Framework ajuda a criar aplicativos escalonáveis, extensíveis, de fácil manutenção e irá nos ajudar a implementar o padrão arquitetural CQRS.

Isso é feito fornecendo implementações dos blocos de construção mais importantes, como agregados, repositórios e barramentos de eventos (RabbitMQ, Kafka, etc). Além disso, ele fornece suporte a anotações, o que permite que você crie agregados (conjunto de eventos que formam o estado atual de um objeto) e ouvintes de eventos sem vincular seu código à lógica específica do framework. Isso permite que você se concentre em suas lógicas de negócio, em vez de na arquitetura, além de tornar seu código mais fácil de testar de forma isolada.

Ele não tenta, de forma alguma, ocultar a arquitetura do CQRS ou qualquer um de seus componentes dos desenvolvedores. No entanto, ajuda quando se trata de garantir a entrega de eventos para seus ouvintes e processa-los na ordem correta.

Além disso, ainda te proporciona uma infraestrutura de testes com uma abordagem que segue as ideias do Behaviour-Driven Development (BDD). Conseguimos focar nossos testes nos eventos publicados em reação ao envio de comandos, evitando assim, uma dependência sobre a implementação da API.

Chega de conversa, Hands On!

Requisitos do projeto:

• Java 10
• Maven 3.3+
• Docker
• Postgres
• RabbitMQ

Barramento de Eventos (RabbitMQ)

Nossas aplicações vão se comunicar através de um barramento de eventos, onde estamos utilizando o RabbitMQ. Se preferir, você pode instalá-lo em sua maquina, porém uma abordagem mais rápida é executar o comando abaixo para subir uma instância do RabbitMQ com o Docker.

docker run -d --hostname my-rabbit --name rabbbit-local -p 15672:15672 -p 5672:5672 rabbitmq:3-management

Estamos exportando duas portas, a 15672 para acessar o painel de controle e a 5672 que será usada por nossas aplicações para se conectar ao RabbitMQ. Acesse o endereçohttp://localhost:15672 e deverá ser apresentado o painel de controle do RabbitMQ. Utilize o usuário guest e senha guest para fazer login e observar a fila bank-account.events sendo populada e consumida.

Aplicação Command

Nossa aplicação command irá se chamarbank-account-command. Crie uma aplicação base Spring Boot e faça com que o seu pom.xml fique parecido com este.

Vamos passar um pouco pelas dependências do projeto que fazem sentido para essa Story, para entendermos melhor qual a importância de cada uma.

• Spring Boot Starter AMQP: Lib fornecida pelo Spring para nos conectarmos facilmente a uma instância do RabbitMQ.
• Axon Spring Boot Starter: Lib fornecida pelo Axon Framework para que aproveitemos o sistema de Auto Configuração do Spring Boot. Ela irá configurar automaticamente os componentes básicos da infraestrutura (Barramento de Comando, Barramento de Eventos), assim como qualquer componente necessário para executar e armazenar Agregados.
• Axon AMQP: Lib do Axon Framework para que possamos configurar o RabbitMQ como nosso barramento de eventos.
• Lombok: O Lombok nos fornece anotações para evitar códigos repetitivos como getters, setters, construtores, builders deixando o código mais limpo e organizado.

Noapplication.yml vamos configurar apenas um nome para nosso canal e fila do RabbitMQ, a porta da nossa aplicação e o datasource para que nossa aplicação utilize o Postgres como Banco de Dados.

O Axon Framework funciona baseado em comandos. Comandos são intenções de alteração de estado que carregam consigo as informações necessárias para realizar a mesma, eles são registrados em um barramento e podem ser executados de forma sincrona ou assincrona.

Precisamos configurar a aplicação para publicar eventos no RabbitMQ, não iremos entrar em detalhes desta parte. Se quiser saber mais acesse a documentação oficial do Spring Boot Starter AMQP.

Nossa aplicação terá três comandos. Adicionar, Remover e Atualizar Saldo da Conta Bancária. Crie as classes abaixo para representar esses comandos.

Note que estamos utilizando a anotação @TargetAggregateIdentifier do Axon. Ela marca o identificador do agregado que o comando visa alterar.

Esses comandos depois de processados resultaram em eventos lançados no RabbitMQ (nosso barramento de eventos). Ressalto que um comando pode se desdobrar em vários eventos depois de processado, porém em nosso exemplo cada comando resultará em um único evento.

Portanto teremos três eventos. Conta Adicionada, Conta Removida e Saldo Atualizado. Crie as classes abaixo para representar esses eventos.

Agora que já temos nossos comandos e eventos vamos criar nosso Agregado. Ele será responsável por tratar nossos comandos, aplicar as regras de negócio e enviar o evento para nosso barramento de eventos (RabbitMQ).

Crie a classe BankAccountAggregate para representar nosso agregado.

Aqui temos varias coisas a serem esclarecidas, vamos começar pelas anotações do Axon.

• @Aggregate: Usada para representar um agregado. Um agregado é uma árvore isolada de entidades que é capaz de manipular comandos. Quando um comando é despachado para um agregado, o Axon carrega a instância agregada e invoca o método manipulador do comando relacionado.
• @AggregateIdentifier: Campo usado como identificador do agregado. O Axon irá utilizar esse campo para buscar todos os eventos relacionados ao agregado e carregar seu estado atual.
• @CommandHandler: Usada para cadastrar um método ou construtor como manipulador de um comando no barramento de comandos do Axon. O nome do método não faz diferença, porém o mesmo deve receber como parâmetro o comando que será manipulado por ele. Usamos esta anotação em um construtor quando com comando manipulado resultará na criação de um agregado.
• @EventSourcingHandler: Anotação que marca um método em um agregado como um manipulador para eventos gerados por esse agregado. É usada para ler todos os eventos e montar o estado atual do agregado quando um comando é recebido por exemplo.

Neste exemplo criamos em nosso agregado um manipulador para cada comando. Não há muito o que dizer aqui, porque nosso exemplo não tem nenhuma regra específica visando a simplicidade da Story, simplesmente recebemos o comando do barramento de comandos, validamos as informações e utilizamos o método estático apply da classeAggregateLifecycle do Axon para enviar os eventos para o nosso barramento de eventos (RabbitMQ).

Agora precisamos de um ponto de entrada que dispare nossos comandos. Vamos criar uma simples API REST que receba as informações em um DTO e envie os comandos para o barramento de comandos.

Crie uma classe chamada BankAccountDTO para receber as informações que serão passadas no body de nossos endpoints.

E enfim criamos o controller que disponibilizará nossa API. Temos algumas particularidades aqui e vamos falar delas a seguir.

Bom, a primeira coisa a se falar é que injetamos uma classe CommandGateway. Usaremos essa instância para enviar comandos para o Command Bus do Axon e ele será responsável por fazer com que os comandos cheguem até os handlers do nosso Agregado.

Como estamos utilizando o método send do CommandGateway , que é assíncrono, temos que retornar um CompletableFuture fazendo com que nossa API seja reativa, se for uma necessidade para sua aplicação trabalhar de forma síncrona, não se preocupe, o CommandGateway também provê um método sendAndWait.

Para ter uma visão geral dos eventos que foram gerados para um agregado crie um controller que faça a consulta no Event Store.

Aplicação Query

A aplicação query será responsável por consumir o barramento de eventos e montar uma projeção em um banco relacional. Vamos chama-lá de bank-account-query. O pom.xml ficará bem parecido com o da aplicação command.

Agora vamos utilizar o application.yml para configurar a porta da aplicação, o acesso ao banco de dados onde será criada a tabela bank_account e o nome da fila que será consumida do RabbitMQ assim como o método que implementará o handler da mesma.

Do mesmo jeito que na aplicação bank-account-command configuramos para publicar eventos no RabbitMQ, precisamos configurar o bank-account-query para ler os eventos de lá.

Para que essa Story não se estenda muito, não vou falar muito da API tradicional de contas bancárias, apenas crie um model, um repository e um controller para que possamos ver o resultado gerado pelo processamento dos eventos.

Agora vamos para a parte que interessa, o processamento dos eventos. Em geral vamos le-los do RabbitMQ e realizar as atualizações necessárias em nosso banco relacional.

Temos algumas anotações novas do Axon aqui, vamos percorrer por elas para entender.

• @ProcessingGroup: Usamos essa anotação para especificar que os eventos serão consumidos do RabbitMQ.
• @EventHandler: Anotação usada para especificar um método manipulador de evento. O método deve receber como parâmetro o evento que deseja escutar.

Quanto as ações executas, acredito que seja bem familiar pra você. Para o BankAccountAddedEvent adicionamos uma nova conta bancária, para o BankAccountRemovedEvent removemos a conta e para BankAccountBalanceUpdatedEvent atualizamos o saldo da conta.

Bom é muita informação, e ao longo do caminho posso ter deixado algo de fora. Você pode consultar o código completo das duas aplicação no GitHub.

Rodando Testando =)

Agora vamos fazer alguns testes para garantir que tudo esteja funcionando. Suba a aplicação command e a query, não se esqueça de que o RabbbitMQ tem que estar rodando.

Crie uma nova conta bancária chamando o seguinte endpoint do bank-account-command:

curl -X POST http://localhost:8081/bank-accounts -H 'Content-Type: application/json' -d '{"name": "Caixa"}'

O retorno deve ser o id da conta criada. Utilize este id para executar as duas requests a seguir que representam as atualizações de saldo.

curl -X PUT http://localhost:8081/bank-accounts/{bankId}/balances -H 'Content-Type: application/json' -d '{"balance": 352.80}'
curl -X PUT http://localhost:8081/bank-accounts/{bankId}/balances -H 'Content-Type: application/json' -d '{"balance": 469.20}'

Busque todas as contas utilizando o endpoint do bank-account-query. Verá que temos uma conta chamadaCaixa cadastrada com o saldo de 469.20.

curl -X GET http://localhost:8082/bank-accounts
[
    {
        "id": "92a43170-5baa-4f04-a5c0-20004eda0d99",
        "name": "Caixa",
        "balance": 469.2
    }
]

Agora consulte o endpoint /events que criamos no bank-account-command para verificar os eventos do nosso agregado e poderemos ver que temos todo o histórico armazenado com varias informações que podem ser bem úteis dependendo da aplicação.

curl -X GET http://localhost:8081/events/{bankId}
[
    {
        "type": "BankAggregate",
        "aggregateIdentifier": "{bankId}",
        "sequenceNumber": 0,
        "identifier": "a09bc3ef-137b-472d-ae26-0994d24780cd",
        "timestamp": "2018-08-20T20:56:47.274121Z",
        "metaData": {
            "traceId": "571ad321-b6ce-4a52-ae52-f48114abf8bb",
            "correlationId": "571ad321-b6ce-4a52-ae52-f48114abf8bb"
        },
        "payload": {
            "id": "92a43170-5baa-4f04-a5c0-20004eda0d99",
            "name": "Caixa",
            "balance": 0
        },
        "payloadType": "br.com.coderef.event.BankAddedEvent"
    },    
    {
        "type": "BankAggregate",
        "aggregateIdentifier": "{bankId}",
        "sequenceNumber": 2,
        "identifier": "8cdaeb49-b206-4cfd-b803-b06d40948019",
        "timestamp": "2018-08-20T21:02:41.932526Z",
        "metaData": {
            "traceId": "e8c4363a-ad80-4abd-95f6-d98edc58eb3e",
            "correlationId": "e8c4363a-ad80-4abd-95f6-d98edc58eb3e"
        },
        "payload": {
            "bankId": "92a43170-5baa-4f04-a5c0-20004eda0d99",
            "balance": 352.8
        },
        "payloadType": "br.com.coderef.event.BankBalanceUpdatedEvent"
    },
    {
        "type": "BankAggregate",
        "aggregateIdentifier": "{bankId}",
        "sequenceNumber": 1,
        "identifier": "540600c2-a5d0-456f-a5e0-114eb618c07e",
        "timestamp": "2018-08-20T20:57:22.629436Z",
        "metaData": {
            "traceId": "ae096401-b7f4-4bc1-a6d8-36ebd44ac2c4",
            "correlationId": "ae096401-b7f4-4bc1-a6d8-36ebd44ac2c4"
        },
        "payload": {
            "bankId": "92a43170-5baa-4f04-a5c0-20004eda0d99",
            "balance": 469.2
        },
        "payloadType": "br.com.coderef.event.BankBalanceUpdatedEvent"
    }
]

Acho que por hoje é isso, é muita informação e com certeza posso ter esquecido de algo. Fique a vontade para deixar seu comentário caso sinta falta de algo. Como disse anteriormente, você pode consultar o código completo da aplicação no GitHub.

Até mais =)

Referências

https://martinfowler.com/bliki/CQRS.html
https://microservices.io/patterns/data/cqrs.html
https://martinfowler.com/eaaDev/EventSourcing.html
https://medium.com/@gabrielqueiroz/vamos-falar-sobre-event-sourcing-276ae66106f7
https://docs.axonframework.org/part-i-getting-started/introduction