A API JMS (Jakarta Messaging 3.1): sessões, mensagens, acknowledgement e seletores

O Jakarta Messaging — que todo mundo ainda chama de JMS — é a API que uma aplicação Java usa para enviar mensagens a um broker e recebê-las dele. A API é um contrato, não um produto: ActiveMQ Classic, ActiveMQ Artemis, IBM MQ, Oracle AQ e muitos outros oferecem implementações, e o código que se limita às interfaces de jakarta.jms migra de uma para outra apenas com mudanças de configuração. Este guia tem como alvo o JMS 3.1 (Jakarta EE 10) e usa o ActiveMQ Classic 6 como broker nos exemplos, porque ele é leve e gratuito para rodar localmente.

Duas APIs, um só modelo

Desde a versão 2.0, o JMS oferece dois estilos de programação.

A API clássica (JMS 1.1) monta uma cadeia de objetos: uma ConnectionFactory cria uma Connection, a conexão cria uma Session e a sessão cria os objetos MessageProducer, MessageConsumer e Message. Ela é verbosa, mas explícita, e é o que os servidores de aplicação usam por baixo dos panos para os message-driven beans.

A API simplificada (a partir do JMS 2.0) junta conexão e sessão em um único JMSContext e oferece os helpers fluentes JMSProducer e JMSConsumer. Tudo o que a API clássica faz, a simplificada faz em menos linhas e com recursos AutoCloseable. Use-a em código novo.

Os dois estilos compartilham os mesmos objetos por baixo, e todos os conceitos a seguir — destinos, tipos de mensagem, acknowledgement, seletores — valem para ambos.

Adicione ao projeto a API e um cliente do broker. O artefato da API fornece as interfaces; o artefato cliente fornece a implementação:

<dependency>
  <groupId>jakarta.jms</groupId>
  <artifactId>jakarta.jms-api</artifactId>
  <version>3.1.0</version>
</dependency>
<dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-client</artifactId>
  <version>6.1.6</version> <!-- any 6.x: Jakarta namespace, Java 17+ -->
</dependency>

Enviando uma mensagem

A ConnectionFactory é o único objeto específico do fornecedor com que você lida. Em uma aplicação standalone, você mesmo a instancia; em um servidor de aplicação, você a obtém via JNDI ou a injeta com @Resource.

import jakarta.jms.ConnectionFactory;
import jakarta.jms.JMSContext;
import jakarta.jms.Queue;
import org.apache.activemq.ActiveMQConnectionFactory;

public class OrderSender {
  public static void main(String[] args) {
    ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

    // JMSContext = Connection + Session; try-with-resources closes both
    try (JMSContext context = factory.createContext()) {
      Queue orders = context.createQueue("orders.placed");
      context.createProducer()
          .setProperty("eventType", "OrderPlaced")   // filterable by selectors
          .setProperty("schemaVersion", 2)
          .setTimeToLive(60_000)                     // expire after 60 s if unconsumed
          .send(orders, "{\"orderId\":\"o-9911\",\"total\":149.90}");
    }
  }
}

Sem argumentos, createContext() usa o modo AUTO_ACKNOWLEDGE; createContext(JMSContext.SESSION_TRANSACTED) devolve um contexto transacional no qual você precisa chamar commit(). createQueue("orders.placed") não cria nada no broker: ele apenas monta, no cliente, a identidade de um destino. O ActiveMQ Classic cria a fila no primeiro uso; outros brokers exigem que um administrador a defina antes, e é por isso que a especificação trata destinos como objetos administrados, normalmente obtidos via JNDI.

O JMSProducer retornado por createProducer() é leve e reutilizável: defina nele o modo de entrega, a prioridade, o tempo de vida e as propriedades, e depois chame send quantas vezes precisar.

Recebendo uma mensagem

Os consumidores têm dois formatos: chamadas bloqueantes a receive e callbacks assíncronos com MessageListener.

try (JMSContext context = factory.createContext()) {
  JMSConsumer consumer = context.createConsumer(context.createQueue("orders.placed"));

  // Blocking receive with a timeout; returns null if nothing arrives in 5 s
  Message message = consumer.receive(5_000);
  if (message != null) {
    String body = message.getBody(String.class);   // JMS 2.0 typed body access
    System.out.println("received " + body);
  }
}

A forma assíncrona registra um listener; a biblioteca cliente chama onMessage na própria thread dela a cada entrega:

JMSContext context = factory.createContext();          // keep open for the listener's lifetime
JMSConsumer consumer = context.createConsumer(context.createQueue("orders.placed"));
consumer.setMessageListener(message -> {
  try {
    process(message.getBody(String.class));
  } catch (JMSException e) {
    throw new RuntimeException(e);                     // makes the broker redeliver
  }
});

Duas regras importam aqui. Uma sessão (e, portanto, um JMSContext) é single-thread: nunca a compartilhe entre threads nem a chame de dentro de um listener que pertença a ela. Além disso, a conexão precisa estar iniciada para que os consumidores recebam qualquer coisa — o JMSContext faz isso automaticamente, enquanto a API clássica exige um connection.start() explícito, que é de longe a causa mais comum do famoso "meu consumidor não recebe nada".

Tipos de mensagem, headers e propriedades

Uma mensagem JMS é composta de headers, propriedades e um corpo tipado. Escolha o tipo de corpo de acordo com o que o consumidor precisa fazer com ele.

Os headers são campos fixos definidos pela especificação. O produtor define alguns (JMSDeliveryMode, JMSPriority, JMSExpiration por meio do tempo de vida, JMSReplyTo, JMSCorrelationID, JMSType e, desde o JMS 2.0, JMSDeliveryTime para entrega agendada); o provedor define o restante (JMSMessageID, JMSTimestamp, JMSDestination, JMSRedelivered). A prioridade vai de 0 a 9, com 4 como padrão, e os brokers só são obrigados a tentar entregar primeiro as prioridades mais altas.

As propriedades são seus próprios pares chave-valor, do tipo booleano, numérico ou String. Propriedades cujo nome começa com JMSX são reservadas: JMSXGroupID e JMSXGroupSeq agrupam mensagens relacionadas para que um único consumidor veja uma sequência inteira em ordem, e JMSXDeliveryCount (obrigatória desde a 2.0) informa ao consumidor quantas vezes ele já viu aquela mensagem — a base para tratar mensagens envenenadas.

TextMessage message = context.createTextMessage(json);
message.setStringProperty("eventType", "OrderPlaced");
message.setStringProperty("JMSXGroupID", "customer-4711");   // keep one customer's messages ordered
message.setJMSType("order");                                  // free-form type hint, rarely used

O modo de entrega merece um comentário à parte. DeliveryMode.PERSISTENT (o padrão) instrui o broker a gravar a mensagem no armazenamento antes de confirmar o envio, de modo que ela sobrevive a um restart do broker; o produtor fica bloqueado até essa confirmação chegar. NON_PERSISTENT pula o disco e a espera, o que é várias vezes mais rápido e adequado para métricas, heartbeats e tudo o que você pode se dar ao luxo de perder.

Modos de acknowledgement: o que é prometido de verdade

Uma mensagem fica na fila até o consumidor confirmá-la. Como essa confirmação acontece é o modo de acknowledgement da sessão, e é ele que define a sua garantia de entrega.

Duas sutilezas pegam muita gente. Primeiro, CLIENT_ACKNOWLEDGE vale para a sessão inteira: confirmar a mensagem 5 também confirma da 1 à 4, então ele não serve para confirmar mensagem por mensagem (para isso, o ActiveMQ oferece um INDIVIDUAL_ACKNOWLEDGE fora do padrão). Segundo, uma sessão transacional agrupa envios e recebimentos: consumir de orders.placed e enviar para invoices.created na mesma transação significa que, para o broker, ou as duas coisas acontecem ou nenhuma. O seu banco de dados não faz parte disso — para isso você precisa de um gerenciador de transações XA via servidor de aplicação, ou do padrão outbox.

try (JMSContext context = factory.createContext(JMSContext.SESSION_TRANSACTED)) {
  JMSConsumer consumer = context.createConsumer(context.createQueue("orders.placed"));
  JMSProducer producer = context.createProducer();
  Message order = consumer.receive();
  try {
    String invoice = createInvoice(order.getBody(String.class));
    producer.send(context.createQueue("invoices.created"), invoice);
    context.commit();          // order removed, invoice visible — atomically
  } catch (RuntimeException e) {
    context.rollback();        // order redelivered, invoice never sent
  }
}

Quando há reentrega, o broker marca JMSRedelivered e incrementa JMSXDeliveryCount. O cliente do ActiveMQ aplica uma política de reentrega (por padrão, seis tentativas com um segundo de intervalo) e depois envia a mensagem para uma dead-letter queue chamada ActiveMQ.DLQ; confira essa fila sempre que mensagens "sumirem".

Seletores

Um seletor de mensagens permite que um consumidor receba apenas as mensagens cujos headers ou propriedades atendem a uma expressão. A sintaxe é um subconjunto das cláusulas WHERE do SQL-92, avaliado pelo broker, de modo que as mensagens que não batem nunca trafegam pela rede até esse consumidor.

JMSConsumer highValue = context.createConsumer(
    context.createQueue("orders.placed"),
    "eventType = 'OrderPlaced' AND total > 1000 AND region IN ('EU', 'UK')");

Os operadores suportados incluem comparações, AND/OR/NOT, BETWEEN, IN, LIKE com os curingas % e _, IS NULL e aritmética. Os identificadores se referem a nomes de propriedades ou aos headers com prefixo JMS que podem ser selecionados (JMSDeliveryMode, JMSPriority, JMSMessageID, JMSTimestamp, JMSCorrelationID, JMSType). Em filas, seletores têm um custo: uma fila cujos únicos consumidores usam seletores que não batem com nada acumula para sempre as mensagens que nenhum deles aceita. Em tópicos eles saem de graça, porque as mensagens que não batem simplesmente não são entregues àquela assinatura.

Tópicos: assinaturas duráveis e compartilhadas

Um consumidor de tópico comum recebe as mensagens publicadas enquanto está conectado, e nada mais. Dois tipos de assinatura mudam isso.

Uma assinatura durável é identificada por um nome (e, na API clássica, por um identificador de cliente). O broker guarda as mensagens dela enquanto o assinante está offline e entrega o acumulado na reconexão:

JMSConsumer audit = context.createDurableConsumer(
    context.createTopic("customer.events"), "audit-service");

Até o JMS 2.0, uma assinatura durável permitia exatamente um consumidor ativo, o que tornava impossível escalar um consumidor de tópico sem truques específicos do provedor, como os virtual topics do ActiveMQ. As assinaturas compartilhadas resolvem isso: vários consumidores usando o mesmo nome de assinatura compartilhada dividem as mensagens entre si, como em uma fila.

// Three instances of the same service can each call this; messages are load-balanced
JMSConsumer worker = context.createSharedDurableConsumer(
    context.createTopic("customer.events"), "email-service");

Use createSharedConsumer quando a assinatura não precisar sobreviver a uma desconexão e createSharedDurableConsumer quando precisar. Lembre-se de chamar unsubscribe(name) ao desativar uma assinatura durável; caso contrário, o broker continua armazenando mensagens para um assinante que nunca vai voltar, e o disco enche.

Request-reply sobre filas

O JMS não tem uma primitiva nativa de chamada remota, mas os headers facilitam construir uma robusta. O solicitante cria uma fila temporária (um destino que só existe enquanto durar a conexão dele), envia a requisição com JMSReplyTo apontando para ela e um JMSCorrelationID novo, e então aguarda uma resposta cujo identificador de correlação corresponda.

try (JMSContext context = factory.createContext()) {
  TemporaryQueue replies = context.createTemporaryQueue();
  String correlationId = UUID.randomUUID().toString();

  context.createProducer()
      .setJMSReplyTo(replies)
      .setJMSCorrelationID(correlationId)
      .send(context.createQueue("pricing.requests"), "{\"sku\":\"A-100\"}");

  // Selector ensures we only pick up our own answer if the temp queue is shared
  Message reply = context.createConsumer(replies, "JMSCorrelationID = '" + correlationId + "'")
      .receive(2_000);
  if (reply == null) throw new IllegalStateException("pricing service did not answer in time");
}

Quem responde lê getJMSReplyTo() e getJMSCorrelationID() da requisição e envia a resposta para esse destino com o mesmo identificador de correlação. Dê um timeout a toda requisição: um solicitante esperando sem timeout é um vazamento de threads só esperando o primeiro soluço do broker.

JMS no Spring: JmsTemplate e @JmsListener

A maioria dos times Java nunca chama a API pura, porque o JmsTemplate do Spring cuida da conexão, da sessão e da liberação de recursos, e o @JmsListener transforma qualquer método em um consumidor apoiado por um pool de threads de escuta. Com o Spring Boot 3.2 ou posterior, o spring-boot-starter-activemq autoconfigura uma ConnectionFactory a partir de spring.activemq.broker-url e a envolve em uma factory com cache, para que cada send não abra uma nova conexão TCP.

@Service
public class OrderEvents {
  private final JmsTemplate jms;
  OrderEvents(JmsTemplate jms) { this.jms = jms; }

  public void orderPlaced(String json) {
    jms.convertAndSend("orders.placed", json, m -> {
      m.setStringProperty("eventType", "OrderPlaced");
      return m;
    });
  }
}

@Component
public class OrderConsumer {
  @JmsListener(destination = "orders.placed", selector = "eventType = 'OrderPlaced'", concurrency = "3-10")
  public void handle(String json, @Header("JMSXDeliveryCount") int attempt) {
    // throwing here rolls back / rejects the message and triggers redelivery
  }
}

Defina spring.jms.listener.session.transacted=true (o padrão do Boot para o container de listeners) para que uma exceção em um método listener faça rollback da mensagem em vez de perdê-la, e defina spring.jms.pub-sub-domain=true somente se os nomes de destino se referirem a tópicos.

Erros comuns

• Esquecer o connection.start() com a API clássica. O consumidor conecta, a fila tem mensagens e nada chega. O JMSContext inicia a conexão de forma preguiçosa ao criar o primeiro consumidor, o que é mais um motivo para preferi-lo.
• Criar uma conexão por mensagem. Conexões são sessões TCP caras, com threads próprias. Crie uma única ConnectionFactory e um pequeno número de conexões de longa duração; no Spring, use CachingConnectionFactory ou o padrão do Boot.
• Compartilhar uma sessão entre threads. Sessões não são thread-safe. Cada thread de trabalho tem a sua própria sessão (ou o seu próprio JMSContext); compartilhe a conexão, não a sessão.
• Usar ObjectMessage. Ele obriga os dois lados a terem a mesma classe no classpath e abre espaço para vulnerabilidades de desserialização Java. Envie JSON em um TextMessage; use o formatador de JSON para inspecionar os payloads durante a depuração.
• Seletores que nunca batem em uma fila. As mensagens se acumulam. Roteie para filas diferentes já no envio ou garanta que algum consumidor aceite tudo.
• Ignorar a dead-letter queue. Mensagens envenenadas vão parar silenciosamente em ActiveMQ.DLQ (ou no equivalente do seu provedor). Monitore essa fila e tenha uma forma de inspecionar e reprocessar o conteúdo dela.

Principais pontos

• JMS é uma API; o cliente do broker que você coloca no classpath decide qual protocolo de rede é usado e qual broker você consegue acessar.
• Prefira JMSContext e try-with-resources; mantenha as sessões em uma única thread e as conexões de longa duração.
• Envie JSON em TextMessage, coloque os metadados de roteamento em propriedades e use JMSXGroupID quando um conjunto de mensagens precisar manter a ordem.
• O modo de acknowledgement define a sua garantia: AUTO e CLIENT dão at-least-once, DUPS_OK troca duplicatas por velocidade, e sessões transacionais tornam atômico no broker o par consumo-produção.
• Seletores filtram apenas por headers e propriedades; assinaturas duráveis compartilhadas permitem escalar consumidores de tópicos como se fossem uma fila.
• Request-reply é fila temporária mais JMSReplyTo mais JMSCorrelationID, sempre com timeout.
• No Spring, JmsTemplate e @JmsListener eliminam o boilerplate sem mudar nenhuma das garantias acima.