Message-driven beans e Jakarta Messaging (JMS 3.1) no contêiner

A maior parte do que você precisa saber sobre mensageria (por que filas desacoplam sistemas, o que um broker garante, como topics diferem de filas) não depende do Jakarta EE e é tratado no tema de mensageria. O que a plataforma acrescenta é uma forma de consumir e produzir mensagens sem gerenciar você mesmo conexões, sessões, threads ou confirmações: o message-driven bean do lado consumidor e um JMSContext injetado do lado produtor, ambos ligados ao gerenciador de transações do contêiner. Essa integração é poderosa e também a origem da maioria das surpresas, por isso este guia se concentra no comportamento do contêiner e não na superfície da API.

O lado consumidor: message-driven beans

Um message-driven bean (MDB) é uma classe anotada com @MessageDriven que implementa jakarta.jms.MessageListener. O contêiner se conecta ao broker por você, cria um pool de instâncias e chama onMessage em uma delas para cada mensagem que chega.

package shop.messaging;

import jakarta.ejb.ActivationConfigProperty;
import jakarta.ejb.MessageDriven;
import jakarta.inject.Inject;
import jakarta.jms.JMSException;
import jakarta.jms.Message;
import jakarta.jms.MessageListener;
import jakarta.jms.TextMessage;

@MessageDriven(activationConfig = {
    @ActivationConfigProperty(propertyName = "destinationLookup", propertyValue = "java:/jms/queue/orders"),
    @ActivationConfigProperty(propertyName = "destinationType",   propertyValue = "jakarta.jms.Queue"),
    @ActivationConfigProperty(propertyName = "messageSelector",   propertyValue = "region = 'EU'")
})
public class OrderPlacedListener implements MessageListener {

    @Inject
    OrderProcessor processor;              // the business logic lives in a testable bean

    @Override
    public void onMessage(Message message) {
        try {
            String payload = message.getBody(String.class);     // JMS 2.0+: typed body access
            String orderId = message.getStringProperty("orderId");
            processor.process(orderId, payload);
        } catch (JMSException e) {
            throw new IllegalStateException("cannot read message", e);   // roll back and redeliver
        }
    }
}

onMessage não tem valor de retorno nem exceções checked; a única forma de sinalizar falha é lançar uma exceção runtime ou marcar a transação para rollback. A configuração de ativação diz ao contêiner o que consumir. As propriedades definidas pela especificação do Messaging são:

Todo o resto (tamanho do pool, atraso de reentrega, tamanho de lote) é uma propriedade proprietária do resource adapter que conecta o servidor ao broker. Isso importa: formalmente, um MDB é um endpoint da arquitetura Jakarta Connectors, e a mesma anotação funciona com adapters para sistemas que não são JMS (e-mail, arquivos, Kafka), desde que você implemente a interface listener do adapter em vez de MessageListener. Quando um servidor tem mais de um adapter, o que um MDB usa é escolhido pelos padrões do servidor ou indicado no descritor de implantação proprietário (por exemplo, jboss-ejb3.xml no WildFly).

O ideal é definir os destinos no servidor, mas para testes e implantações autocontidas você pode declará-los no código:

@JMSDestinationDefinition(
    name = "java:/jms/queue/orders",
    interfaceName = "jakarta.jms.Queue",
    destinationName = "orders")
@JMSConnectionFactoryDefinition(
    name = "java:/jms/OrdersCF",
    transactional = true)
@ApplicationScoped
public class MessagingResources { }

O lado produtor: JMSContext

O JMS 1.1 exigia uma Connection, uma Session, um MessageProducer e um try/finally para fechar tudo. O JMS 2.0 (Java EE 7) introduziu a API simplificada: JMSContext encapsula conexão e sessão, JMSProducer é um builder fluente e, dentro de um contêiner, você injeta o contexto e nunca o fecha.

import jakarta.annotation.Resource;
import jakarta.ejb.Stateless;
import jakarta.inject.Inject;
import jakarta.jms.*;

@Stateless
public class OrderPublisher {

    @Inject
    @JMSConnectionFactory("java:/jms/OrdersCF")   // optional; default factory if omitted
    JMSContext context;                            // container-managed, scoped to the transaction

    @Resource(lookup = "java:/jms/queue/orders")
    Queue orders;

    public void orderPlaced(Order order) {
        context.createProducer()
               .setProperty("orderId", order.getId().toString())
               .setProperty("region", order.getRegion())
               .setDeliveryMode(DeliveryMode.PERSISTENT)
               .setTimeToLive(24 * 60 * 60 * 1000L)
               .send(orders, toJson(order));       // send(Destination, String) creates a TextMessage
    }
}

O JMSContext injetado tem escopo de transação: quando a transação JTA do session bean faz commit, a mensagem é entregue; quando sofre rollback, a mensagem é descartada. Se o mesmo método também grava no banco via JPA, as duas coisas acontecem de forma atômica, desde que a connection factory suporte XA e o gerenciador de transações coordene os dois recursos, que é o assunto do guia de transações. Fora de uma transação, o contexto se comporta como uma sessão em auto-acknowledge e as mensagens saem imediatamente.

Para o corpo das mensagens, prefira TextMessage com JSON ou BytesMessage com uma codificação documentada. ObjectMessage com serialização Java amarra produtor e consumidor às mesmas versões de classe e é um vetor de ataques de desserialização; cada vez mais brokers e servidores o bloqueiam por padrão. MapMessage serve para payloads pequenos e planos.

Como as transações determinam a confirmação

Esta é a parte que confunde quem aprendeu JMS fora de um contêiner. Em um cliente standalone, você escolhe um modo de confirmação (auto, client ou dups-ok) ou uma sessão transacional, e chama acknowledge() ou commit() por conta própria. Em um MDB com transações gerenciadas pelo contêiner, que é o padrão, nada disso se aplica:

1. O contêiner recebe a mensagem e inicia uma transação JTA.
2. Ele inscreve a sessão JMS nessa transação e chama onMessage.
3. Qualquer trabalho com JPA, outros envios JMS ou chamadas JDBC no método também entram na transação.
4. Se onMessage retorna normalmente, o contêiner faz commit: as alterações no banco são gravadas, as mensagens de saída são liberadas e a mensagem de entrada é confirmada, tudo junto.
5. Se onMessage lança uma exceção runtime ou chama messageDrivenContext.setRollbackOnly(), tudo sofre rollback e o broker reentrega a mensagem, com JMSRedelivered igual a true e JMSXDeliveryCount incrementado.

O atributo de transação do MDB pode ser REQUIRED (o padrão) ou NOT_SUPPORTED; os outros não fazem sentido porque não existe transação do chamador à qual se juntar. Com @TransactionManagement(BEAN), a mensagem é confirmada de acordo com acknowledgeMode quando onMessage retorna, independentemente do que a sua UserTransaction fez; assim, um MDB com transações gerenciadas pelo bean pode fazer commit do trabalho no banco e ainda perder a confirmação, ou confirmar uma mensagem cujo processamento falhou. Fique com as transações gerenciadas pelo contêiner, a menos que tenha um motivo específico.

Filas, topics e assinaturas

Uma fila entrega cada mensagem a um consumidor; um topic entrega cada mensagem a todos os assinantes. Dois detalhes do consumo de topics são específicos do contêiner:

• Uma assinatura durável sobrevive à remoção do deploy do MDB ou à queda do servidor: o broker guarda as mensagens até o assinante voltar. Ela é identificada por clientId mais subscriptionName, e uma assinatura durável no estilo do JMS 1.1 só pode ter um consumidor ativo. Isso conflita com um pool de MDB, e é por isso que o JMS 2.0 acrescentou as assinaturas compartilhadas à API (createSharedDurableConsumer); se um pool de MDB usa uma delas depende de uma propriedade de ativação do fornecedor (o adapter do Artemis no WildFly a chama de shareSubscriptions), e com ela várias instâncias, mesmo em vários nós de um cluster, compartilham uma única assinatura durável.
• Uma assinatura não durável de um topic perde as mensagens publicadas enquanto o MDB não está rodando. Isso serve para broadcasts de invalidação de cache e é um erro para qualquer coisa crítica para o negócio.

Seletores de mensagem permitem que um consumidor receba um subconjunto das mensagens de um destino com base em cabeçalhos e propriedades (region = 'EU' AND priority > 4). Eles são avaliados pelo broker, então mensagens que não batem nunca são enviadas ao consumidor; mas, em uma fila, elas também nunca são consumidas por este consumidor, e se nenhum consumidor as selecionar ficam lá para sempre. Use seletores à vontade em topics, e em filas só quando toda mensagem tiver garantido um consumidor correspondente.

Concorrência, pooling e ordem

O contêiner executa várias instâncias de MDB em paralelo (o pool padrão de MDBs do WildFly e o maxConcurrency do Open Liberty são ajustes típicos) e cada instância recebe mensagens de forma independente. A concorrência é o que faz os MDBs escalarem, e ela tem duas consequências:

• A ordem não é preservada entre instâncias. Duas mensagens do mesmo pedido podem ser processadas em paralelo e terminar na sequência errada. Reduza o pool a uma instância para ordem estrita, use message groups do broker (JMSXGroupID no ActiveMQ Artemis) para prender uma chave a um único consumidor, ou torne o processamento independente da ordem levando uma versão ou um timestamp na mensagem.
• Um onMessage lento ocupa uma vaga do pool. Trabalho demorado dentro de um MDB deixa outras mensagens esperando; se o processamento é lento, confirme rápido repassando o trabalho para uma fila própria ou para um managed executor, dentro dos limites transacionais que você puder aceitar.

Do ponto de vista do contêiner, as instâncias não têm estado: assim como os stateless session beans, ficam em pool, e seus campos devem guardar apenas recursos injetados.

Projetando um consumidor robusto

Como a reentrega existe, todo consumidor precisa ser idempotente: processar a mesma mensagem duas vezes deve ter o mesmo efeito que processá-la uma vez. Técnicas práticas, da mais barata à mais robusta:

1. Tornar a operação naturalmente idempotente: definir um status em vez de incrementar um contador.
2. Gravar o JMSMessageID (ou uma chave de negócio enviada como propriedade) na mesma transação de banco que o efeito, com uma restrição de unicidade; uma duplicata faz o insert falhar e você faz commit sem repetir o trabalho.
3. Para efeitos colaterais externos que não podem ser desfeitos (enviar e-mail, chamar uma API de pagamento), registrar primeiro a intenção no banco e executar o efeito a partir de um consumidor separado que verifica esse registro.

Valide a mensagem antes de tocar em qualquer recurso e envie as mensagens que nunca poderão ser processadas (schema inválido, tipo desconhecido) para uma fila de erro com o motivo em uma propriedade; depois retorne normalmente, para que o broker pare de reentregá-las. Registre o JMSMessageID e o id de correlação em toda linha de log: quando uma mensagem fica presa em reentrega, é por esse id que você a encontra no console do broker. E mantenha o MDB enxuto (parsing mais uma chamada a um service bean), para que a lógica possa ser testada com um teste JUnit simples e o MDB uma única vez com um broker embutido.

O que o Jakarta Messaging 3.1 muda e o que não muda

O Messaging 3.0 foi a troca de namespace (javax.jms para jakarta.jms); o 3.1, parte do Jakarta EE 10, elevou a versão mínima do Java e fez pequenos esclarecimentos, mas não trouxe nenhuma API nova. A próxima geração discutida há tempos (listeners nativos de CDI em beans comuns, uma API reativa) não saiu, então o MDB continua sendo a única forma padrão e portável de consumir mensagens em um servidor Jakarta EE. Os fornecedores preenchem a lacuna: WildFly e Open Liberty expõem recursos do Artemis e do próprio motor por meio de propriedades de ativação, e o MicroProfile Reactive Messaging oferece uma alternativa baseada em anotações nos runtimes que o suportam. Quando você ler código com javax.jms, @MessageDriven de javax.ejb e destinationLookup, é o mesmo modelo; só o prefixo mudou.

Erros comuns

• Lançar exceção em falhas permanentes. A mensagem entra em loop de reentrega até o limite do broker. Encaminhe para uma fila de erro e retorne normalmente.
• Processamento não idempotente. Uma reentrega após uma queda entre o commit e a confirmação, rara mas real, processa a mensagem duas vezes. Deduplique pelo id da mensagem ou por uma chave de negócio.
• Transações gerenciadas pelo bean em um MDB. A confirmação e a sua transação ficam desacopladas; você pode perder mensagens ou processá-las duas vezes. Use transações gerenciadas pelo contêiner.
• Seletores em filas sem um consumidor que pegue o resto. Mensagens que não batem se acumulam silenciosamente. Monitore a profundidade da fila ou use um topic.
• Presumir processamento ordenado. O pool processa em paralelo. Projete levando isso em conta ou restrinja o pool.
• ObjectMessage com classes de domínio. Acoplamento de versões e risco de desserialização. Use JSON em um TextMessage.
• Fechar um JMSContext injetado. Ele pertence ao contêiner; fechá-lo lança exceção ou quebra chamadas posteriores. Crie seu próprio contexto a partir de uma ConnectionFactory somente em código que roda fora do gerenciamento do contêiner.

Principais pontos

• Um message-driven bean transforma uma fila ou um topic em uma chamada de método: o contêiner gerencia a conexão, um pool de instâncias e a transação em torno de cada onMessage.
• As propriedades de ativação (destinationLookup, destinationType, messageSelector, configurações de assinatura) são padrão; tamanho do pool, limites de reentrega e tratamento de dead letters são configuração do broker e do servidor.
• Com transações gerenciadas pelo contêiner, a confirmação faz parte do commit: um retorno normal confirma, e uma exceção faz rollback e dispara a reentrega.
• Injete JMSContext para enviar; ele participa da transação JTA em andamento, de modo que gravações no banco e mensagens de saída fazem commit juntas.
• Assinaturas duráveis guardam as mensagens de um topic enquanto o consumidor está fora do ar, assinaturas compartilhadas permitem que um pool ou cluster as consuma, e os seletores filtram no broker.
• Os consumidores devem ser idempotentes, validar cedo e desviar mensagens impossíveis de processar em vez de lançar exceções para sempre.
• O Messaging 3.1 não mudou nada funcionalmente; o modelo de MDB é o mesmo do Java EE 7, com pacotes jakarta.*.