Usando o padrão observer com CDI para lidar com o Kafka

Talvez haja mais servidores de mensageria do que de aplicação JavaEE, então o analista tem muitos trade-offs para avaliar. De qualquer forma, mesmo que futuramente haja uma troca para um servidor JMS, as partes dos sistema que usam esse serviço não devem ser alteradas, aí que está o desafio do acoplamento.

The more coupling we have between objects, components, modules, or systems, the more we experience many consequences. These consequences include but are not limited to difficulty of modification, propagation of failure, inability to scale because of contention, and performance issues due to dependent actions.

Encapsulando a API.

A API do Kafka é bem tranquila de usar, baseada em Producer e Consumer, portanto o primeiro desafio era encapsular o uso em uma interface mais simples ainda. Outro problema aparece, o encapsulamento. Nenhuma das partes do sistema devem saber sobre os detalhes de como foi implementado o serviço de mensageria, devem apenas saber o mínimo possível para utilizá-lo.

Da interface acima, poderíamos diminuir ainda mais, como omitir método assinar, identificando que o contexto da aplicação foi inicializado e fazendo a assinatura automática.

Agora basta usar um framework de injeção de dependências para usar o serviço de mensageria, sem saber quem é o responsável por prover a implementação.

O padrão Observer

Agora passamos para o próximo problema, o cliente da API já fez a assinatura de um fila de mensagens e precisa ser notificado quando houver alguma mensagem na fila. Para isso, nada mais adequado do que implementar o padrão Observer. Ao invés de implementar o padrão, usei uma api que já implementa, inclusive discutida pelo Sérgio Lopes anteriormente, o CDI.

Lidando com o sincronismo.

Agora precisamos retirar mensagens da fila e produzir eventos do CDI. Só que aqui temos um sério problema: a API de Consumer do Kafka fica em um loop infinito while true e caso haja mensagem um evento deve ser disparado.

O problema aqui é que essa solução não escala e será muito lenta, devido a basicamente um limitador: o CDI só trabalha com eventos síncronos, ou seja, quando a Thread chega na linha evento.fire(), ela só passa para o próximo passo quando todos os responsáveis por ouvir o evento @Observer SerieMensagens terminam o seu processamento. Há como contornar esse "problema", tanto no produtor do evento como no observador do evento. Neste, poderíamos utilizar o Asynchonous e naquele bastaria lançar o evento em uma thread separada.

Concluindo.

Pensar em baixo acoplamento, alta coesão e encapsulamento é problema diário de qualquer desenvolvedor. Você pode até não perceber de imediato, mas o preço da manutenibilidade chega logo (Design payoff). Por fim, há várias outras features do CDI que você deveria ter no seu leque de opções e talvez no futuro, a invocação de métodos assíncronos faça parte da especificação. Se bem que o JavaEE 8 não está tão perto assim de sair, no momento passa por uma crise interna.