La API JMS (Jakarta Messaging 3.1): sesiones, mensajes, acuse de recibo y selectores

Jakarta Messaging, al que todo el mundo sigue llamando JMS, es la API que usa una aplicación Java para enviar mensajes a un broker y recibirlos de él. La API es un contrato, no un producto: ActiveMQ Classic, ActiveMQ Artemis, IBM MQ, Oracle AQ y muchos otros ofrecen implementaciones, y el código que se limita a las interfaces de jakarta.jms pasa de una a otra cambiando solo la configuración. Esta guía se centra en JMS 3.1 (Jakarta EE 10) y usa ActiveMQ Classic 6 como broker en los ejemplos, porque es ligero y gratuito de ejecutar en local.

Dos API, un mismo modelo

Desde la versión 2.0, JMS ofrece dos estilos de programación.

La API clásica (JMS 1.1) construye una cadena de objetos: una ConnectionFactory crea una Connection, la conexión crea una Session y la sesión crea los objetos MessageProducer, MessageConsumer y Message. Es verbosa pero explícita, y es la que usan internamente los servidores de aplicaciones para los message-driven beans.

La API simplificada (desde JMS 2.0) reúne conexión y sesión en un único JMSContext y le añade los helpers fluidos JMSProducer y JMSConsumer. Todo lo que hace la API clásica lo hace la simplificada en menos líneas y con recursos AutoCloseable. Es la que debes usar en código nuevo.

Ambos estilos comparten los mismos objetos por debajo, y todos los conceptos que siguen —destinos, tipos de mensaje, acuse de recibo, selectores— se aplican a los dos.

Añade al proyecto la API y un cliente del broker. El artefacto de la API aporta las interfaces; el artefacto cliente, la implementación:

<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>

Enviar un mensaje

La ConnectionFactory es el único objeto específico del proveedor que vas a tocar. En una aplicación standalone la construyes tú; en un servidor de aplicaciones la obtienes por JNDI o la inyectas con @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}");
    }
  }
}

Sin argumentos, createContext() te da el modo AUTO_ACKNOWLEDGE; createContext(JMSContext.SESSION_TRANSACTED) te da un contexto transaccional en el que debes llamar a commit(). createQueue("orders.placed") no crea nada en el broker: construye en el cliente la identidad de un destino. ActiveMQ Classic crea la cola la primera vez que se usa; otros brokers exigen que un administrador la haya definido antes, y por eso la especificación trata los destinos como objetos administrados que normalmente se obtienen de JNDI.

El JMSProducer que devuelve createProducer() es ligero y reutilizable: configura en él el modo de entrega, la prioridad, el tiempo de vida y las propiedades, y luego llama a send tantas veces como necesites.

Recibir un mensaje

Los consumidores tienen dos formas: llamadas bloqueantes a receive y callbacks asíncronos con 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);
  }
}

La forma asíncrona registra un listener; la biblioteca cliente llama a onMessage desde su propio thread en 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
  }
});

Aquí importan dos reglas. Una sesión (y, por tanto, un JMSContext) es de un solo thread: nunca la compartas entre threads ni la llames desde dentro de un listener que le pertenezca. Además, la conexión tiene que estar arrancada para que los consumidores reciban algo: JMSContext lo hace automáticamente, mientras que la API clásica exige un connection.start() explícito, que es con diferencia la causa más habitual del clásico «mi consumidor no recibe nada».

Tipos de mensaje, cabeceras y propiedades

Un mensaje JMS está formado por cabeceras, propiedades y un cuerpo tipado. Elige el tipo de cuerpo según lo que el consumidor tenga que hacer con él.

Las cabeceras son campos fijos definidos por la especificación. El productor fija algunas (JMSDeliveryMode, JMSPriority, JMSExpiration a través del tiempo de vida, JMSReplyTo, JMSCorrelationID, JMSType y, desde JMS 2.0, JMSDeliveryTime para la entrega programada); el proveedor fija el resto (JMSMessageID, JMSTimestamp, JMSDestination, JMSRedelivered). La prioridad va de 0 a 9, con 4 por defecto, y los brokers solo están obligados a intentar entregar primero las prioridades más altas.

Las propiedades son tus propios pares clave-valor, de tipo booleano, numérico o String. Las propiedades cuyo nombre empieza por JMSX están reservadas: JMSXGroupID y JMSXGroupSeq agrupan mensajes relacionados para que un solo consumidor vea una secuencia completa en orden, y JMSXDeliveryCount (obligatoria desde la 2.0) indica al consumidor cuántas veces ha visto ya ese mensaje, que es la base para gestionar los mensajes envenenados.

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

El modo de entrega merece un comentario aparte. DeliveryMode.PERSISTENT (el valor por defecto) indica al broker que escriba el mensaje en su almacenamiento antes de confirmar el envío, de modo que el mensaje sobrevive a un reinicio del broker; el productor se queda bloqueado hasta recibir esa confirmación. NON_PERSISTENT se salta el disco y la espera: es varias veces más rápido y adecuado para métricas, heartbeats y todo lo que te puedas permitir perder.

Modos de acuse de recibo: qué se promete de verdad

Un mensaje permanece en la cola hasta que el consumidor lo confirma. Cómo se produce esa confirmación es el modo de acuse de recibo de la sesión, y es lo que determina tu garantía de entrega.

Hay dos sutilezas con las que tropieza mucha gente. La primera: CLIENT_ACKNOWLEDGE afecta a toda la sesión; confirmar el mensaje 5 confirma también del 1 al 4, así que no sirve para confirmar mensaje a mensaje (para eso ActiveMQ ofrece un INDIVIDUAL_ACKNOWLEDGE no estándar). La segunda: una sesión transaccional agrupa envíos y recepciones; consumir de orders.placed y enviar a invoices.created en una misma transacción significa que, para el broker, ocurren las dos cosas o ninguna. Tu base de datos no entra en ella: para eso necesitas un gestor de transacciones XA a través de un servidor de aplicaciones, o el patrón 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
  }
}

Cuando hay una reentrega, el broker marca JMSRedelivered e incrementa JMSXDeliveryCount. El cliente de ActiveMQ aplica una política de reentrega (por defecto, seis intentos con un segundo de espera) y después envía el mensaje a una dead-letter queue llamada ActiveMQ.DLQ; revisa esa cola siempre que haya mensajes que «desaparecen».

Selectores

Un selector de mensajes permite que un consumidor reciba solo los mensajes cuyas cabeceras o propiedades cumplen una expresión. La sintaxis es un subconjunto de las cláusulas WHERE de SQL-92 y la evalúa el broker, de modo que los mensajes que no coinciden nunca viajan por la red hacia ese consumidor.

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

Los operadores admitidos incluyen comparaciones, AND/OR/NOT, BETWEEN, IN, LIKE con los comodines % y _, IS NULL y aritmética. Los identificadores hacen referencia a nombres de propiedades o a las cabeceras con prefijo JMS que se pueden seleccionar (JMSDeliveryMode, JMSPriority, JMSMessageID, JMSTimestamp, JMSCorrelationID, JMSType). En las colas, los selectores tienen un coste: una cola cuyos únicos consumidores usan selectores que no coinciden con nada acumula para siempre los mensajes que no coinciden con ninguno. En los topics salen gratis, porque los mensajes que no coinciden simplemente no se entregan a esa suscripción.

Topics: suscripciones durables y compartidas

Un consumidor de topic normal recibe los mensajes publicados mientras está conectado, y nada más. Hay dos tipos de suscripción que cambian eso.

Una suscripción durable se identifica por un nombre (y, en la API clásica, por un identificador de cliente). El broker guarda sus mensajes mientras el suscriptor está desconectado y le entrega los pendientes al reconectarse:

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

Hasta JMS 2.0, una suscripción durable admitía exactamente un consumidor activo, lo que impedía escalar un consumidor de topic sin trucos específicos del proveedor, como los virtual topics de ActiveMQ. Las suscripciones compartidas lo resuelven: varios consumidores que usan el mismo nombre de suscripción compartida se reparten los mensajes, como en una cola.

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

Usa createSharedConsumer cuando la suscripción no necesite sobrevivir a una desconexión y createSharedDurableConsumer cuando sí. Acuérdate de llamar a unsubscribe(name) cuando retires una suscripción durable; si no, el broker seguirá almacenando mensajes para un suscriptor que no volverá nunca, y el disco acabará llenándose.

Request-reply sobre colas

JMS no tiene una primitiva de llamada remota integrada, pero con las cabeceras es fácil construir una robusta. El solicitante crea una cola temporal (un destino que solo existe mientras dura su conexión), envía la petición con JMSReplyTo apuntando a ella y un JMSCorrelationID nuevo, y espera una respuesta cuyo identificador de correlación coincida.

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");
}

El que responde lee getJMSReplyTo() y getJMSCorrelationID() de la petición y envía su respuesta a ese destino con el mismo identificador de correlación. Pon siempre un timeout a cada petición: un solicitante que espera sin timeout es una fuga de threads a la espera del primer tropiezo del broker.

JMS en Spring: JmsTemplate y @JmsListener

La mayoría de los equipos Java nunca llaman a la API directa, porque el JmsTemplate de Spring se encarga de la conexión, la sesión y la liberación de recursos, y @JmsListener convierte cualquier método en un consumidor respaldado por un pool de threads de escucha. Con Spring Boot 3.2 o posterior, spring-boot-starter-activemq autoconfigura una ConnectionFactory a partir de spring.activemq.broker-url y la envuelve en una factory con caché para que cada send no abra una conexión TCP nueva.

@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
  }
}

Establece spring.jms.listener.session.transacted=true (el valor por defecto de Boot para el contenedor de listeners) para que una excepción en un método listener revierta el mensaje en lugar de perderlo, y establece spring.jms.pub-sub-domain=true solo si los nombres de destino se refieren a topics.

Errores habituales

• Olvidar connection.start() con la API clásica. El consumidor se conecta, la cola tiene mensajes y no llega nada. JMSContext arranca la conexión de forma perezosa al crear el primer consumidor, lo cual es un motivo más para preferirlo.
• Crear una conexión por mensaje. Las conexiones son sesiones TCP costosas con sus propios threads. Crea una sola ConnectionFactory y un número reducido de conexiones de larga duración; en Spring, usa CachingConnectionFactory o la configuración por defecto de Boot.
• Compartir una sesión entre threads. Las sesiones no son thread-safe. Cada thread de trabajo tiene su propia sesión (o su propio JMSContext); comparte la conexión, no la sesión.
• Usar ObjectMessage. Obliga a que ambos lados tengan la misma clase en el classpath y abre la puerta a vulnerabilidades de deserialización de Java. Envía JSON en un TextMessage; usa el formateador JSON para inspeccionar los payloads mientras depuras.
• Selectores que nunca coinciden en una cola. Los mensajes se acumulan. Enruta a colas distintas en el momento del envío o asegúrate de que algún consumidor lo acepte todo.
• Ignorar la dead-letter queue. Los mensajes envenenados acaban en silencio en ActiveMQ.DLQ (o en el equivalente de tu proveedor). Monitorízala y prepara una forma de inspeccionar y reprocesar su contenido.

Ideas clave

• JMS es una API; el cliente del broker que añades al classpath decide qué protocolo de red se usa y a qué broker puedes conectarte.
• Prefiere JMSContext y try-with-resources; mantén las sesiones en un solo thread y las conexiones de larga duración.
• Envía JSON en TextMessage, pon los metadatos de enrutamiento en propiedades y usa JMSXGroupID cuando un conjunto de mensajes deba mantener su orden.
• El modo de acuse define tu garantía: AUTO y CLIENT dan at-least-once, DUPS_OK cambia duplicados por velocidad y las sesiones transaccionales hacen atómico en el broker el conjunto de consumir y producir.
• Los selectores filtran solo por cabeceras y propiedades; las suscripciones durables compartidas permiten escalar los consumidores de topics como si fueran una cola.
• Request-reply es cola temporal más JMSReplyTo más JMSCorrelationID, siempre con timeout.
• En Spring, JmsTemplate y @JmsListener eliminan el código repetitivo sin cambiar ninguna de las garantías anteriores.