Message-driven beans y Jakarta Messaging (JMS 3.1) en el contenedor

La mayor parte de lo que necesitas saber sobre mensajería (por qué las colas desacoplan sistemas, qué garantiza un broker, en qué se diferencian los topics de las colas) no depende de Jakarta EE y se trata en el tema de mensajería. Lo que aporta la plataforma es una forma de consumir y producir mensajes sin gestionar tú mismo conexiones, sesiones, threads ni acuses de recibo: el message-driven bean en el lado consumidor y un JMSContext inyectado en el lado productor, ambos conectados al gestor de transacciones del contenedor. Esa integración es potente y también el origen de la mayoría de las sorpresas, así que esta guía se centra en cómo se comporta el contenedor más que en la superficie de la API.

El lado consumidor: message-driven beans

Un message-driven bean (MDB) es una clase anotada con @MessageDriven que implementa jakarta.jms.MessageListener. El contenedor se conecta al broker por ti, crea un pool de instancias y llama a onMessage en una de ellas por cada mensaje que llega.

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 no devuelve nada ni declara excepciones checked; la única forma de indicar un fallo es lanzar una excepción runtime o marcar la transacción para rollback. La configuración de activación le dice al contenedor qué debe consumir. Las propiedades que define la especificación de Messaging son:

Todo lo demás (tamaño del pool, retardo de reentrega, tamaño de lote) es una propiedad propietaria del resource adapter que conecta el servidor con el broker. Esto importa: formalmente, un MDB es un endpoint de la arquitectura Jakarta Connectors, y la misma anotación funciona con adapters para sistemas que no son JMS (correo, archivos, Kafka) siempre que implementes la interfaz listener del adapter en lugar de MessageListener. Cuando un servidor tiene más de un adapter, el que usa un MDB lo eligen los valores por defecto del servidor o se indica en el descriptor de despliegue propietario (por ejemplo jboss-ejb3.xml en WildFly).

Lo ideal es definir los destinos en el servidor, pero para tests y despliegues autocontenidos puedes declararlos en 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 { }

El lado productor: JMSContext

JMS 1.1 exigía una Connection, una Session, un MessageProducer y un try/finally para cerrarlos todos. JMS 2.0 (Java EE 7) introdujo la API simplificada: JMSContext envuelve conexión y sesión, JMSProducer es un builder fluido y, dentro de un contenedor, inyectas el contexto y nunca lo cierras.

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

El JMSContext inyectado tiene ámbito de transacción: cuando la transacción JTA del session bean se confirma, el mensaje se entrega; cuando se revierte, se descarta. Si el mismo método también escribe en la base de datos mediante JPA, ambas cosas ocurren de forma atómica, siempre que la connection factory soporte XA y el gestor de transacciones coordine ambos recursos, que es el tema de la guía de transacciones. Fuera de una transacción, el contexto se comporta como una sesión en modo auto-acknowledge y los mensajes salen de inmediato.

Para el cuerpo de los mensajes, prefiere TextMessage con JSON o BytesMessage con una codificación documentada. ObjectMessage con serialización Java ata productor y consumidor a las mismas versiones de clases y es un vector de ataques de deserialización; brokers y servidores lo bloquean cada vez más por defecto. MapMessage sirve para payloads pequeños y planos.

Cómo las transacciones determinan el acuse de recibo

Esta es la parte que confunde a quienes aprendieron JMS fuera de un contenedor. En un cliente independiente eliges un modo de acuse de recibo (auto, client o dups-ok) o una sesión transaccional, y llamas tú a acknowledge() o commit(). En un MDB con transacciones gestionadas por el contenedor, que es lo predeterminado, nada de eso se aplica:

1. El contenedor recibe el mensaje e inicia una transacción JTA.
2. Inscribe la sesión JMS en esa transacción y llama a onMessage.
3. Cualquier trabajo con JPA, otros envíos JMS o llamadas JDBC dentro del método se inscriben también.
4. Si onMessage termina con normalidad, el contenedor confirma: los cambios en la base de datos se escriben, los mensajes salientes se liberan y el mensaje entrante se confirma, todo a la vez.
5. Si onMessage lanza una excepción runtime o llama a messageDrivenContext.setRollbackOnly(), todo se revierte y el broker vuelve a entregar el mensaje, con JMSRedelivered a true y JMSXDeliveryCount incrementado.

El atributo transaccional del MDB puede ser REQUIRED (el predeterminado) o NOT_SUPPORTED; los demás no tienen sentido porque no hay transacción del llamante a la que unirse. Con @TransactionManagement(BEAN), el mensaje se confirma según acknowledgeMode cuando onMessage retorna, independientemente de lo que haya hecho tu UserTransaction; así, un MDB con transacciones gestionadas por el bean puede confirmar su trabajo en la base de datos y aun así perder el acuse de recibo, o confirmar un mensaje cuyo procesamiento falló. Quédate con las transacciones gestionadas por el contenedor salvo que tengas un motivo concreto.

Colas, topics y suscripciones

Una cola entrega cada mensaje a un consumidor; un topic entrega cada mensaje a todos los suscriptores. Hay dos detalles del consumo de topics que son propios del contenedor:

• Una suscripción durable sobrevive a que se retire el despliegue del MDB o a que el servidor esté caído: el broker guarda los mensajes hasta que vuelve el suscriptor. Se identifica por clientId más subscriptionName, y una suscripción durable al estilo de JMS 1.1 solo puede tener un consumidor activo. Eso choca con un pool de MDB, y por eso JMS 2.0 añadió las suscripciones compartidas a la API (createSharedDurableConsumer); que un pool de MDB use una depende de una propiedad de activación del proveedor (el adapter de Artemis de WildFly la llama shareSubscriptions), y con ella varias instancias, incluso en varios nodos de un clúster, comparten una misma suscripción durable.
• Una suscripción no durable a un topic pierde los mensajes publicados mientras el MDB no está en ejecución. Eso está bien para difusiones de invalidación de caché y es un error para cualquier cosa crítica para el negocio.

Los selectores de mensajes permiten que un consumidor reciba un subconjunto de los mensajes de un destino en función de cabeceras y propiedades (region = 'EU' AND priority > 4). Los evalúa el broker, así que los mensajes que no coinciden nunca se envían al consumidor; pero en una cola tampoco los consume este consumidor, y si ningún consumidor los selecciona se quedan ahí para siempre. Usa selectores en topics sin problema, y en colas solo cuando cada mensaje tenga garantizado un consumidor que coincida.

Concurrencia, pooling y orden

El contenedor ejecuta varias instancias de MDB a la vez (el pool por defecto de WildFly para MDB y el maxConcurrency de Open Liberty son ajustes típicos) y cada instancia recibe mensajes de forma independiente. La concurrencia es lo que hace que los MDB escalen, y tiene dos consecuencias:

• El orden no se conserva entre instancias. Dos mensajes del mismo pedido pueden procesarse en paralelo y terminar en la secuencia equivocada. Reduce el pool a una instancia si necesitas un orden estricto, usa los message groups del broker (JMSXGroupID en ActiveMQ Artemis) para asociar una clave a un único consumidor, o haz que el procesamiento no dependa del orden incluyendo una versión o una marca de tiempo en el mensaje.
• Un onMessage lento bloquea un hueco del pool. El trabajo de larga duración dentro de un MDB deja sin atender a otros mensajes; si el procesamiento es lento, confirma rápido pasando el trabajo a una cola propia o a un managed executor, dentro de los límites transaccionales que puedas aceptar.

Desde el punto de vista del contenedor, las instancias no tienen estado: igual que los stateless session beans, están en un pool, y sus campos solo deben contener recursos inyectados.

Diseñar un consumidor robusto

Como existe la reentrega, todo consumidor debe ser idempotente: procesar el mismo mensaje dos veces debe tener el mismo efecto que procesarlo una. Técnicas prácticas, de la más barata a la más sólida:

1. Hacer que la operación sea idempotente por naturaleza: fijar un estado en lugar de incrementar un contador.
2. Guardar el JMSMessageID (o una clave de negocio enviada como propiedad) en la misma transacción de base de datos que el efecto, con una restricción de unicidad; un duplicado hará fallar el insert y confirmarás sin repetir el trabajo.
3. Para efectos secundarios externos que no pueden revertirse (enviar un correo, llamar a una API de pagos), registrar primero la intención en la base de datos y ejecutar el efecto desde un consumidor aparte que compruebe ese registro.

Valida el mensaje antes de tocar ningún recurso, y envía los mensajes que nunca podrán procesarse (esquema incorrecto, tipo desconocido) a una cola de errores con el motivo en una propiedad; después, retorna con normalidad para que el broker deje de reentregarlos. Registra el JMSMessageID y el id de correlación en cada línea de log: cuando un mensaje se queda atascado en reentregas, ese id es lo que te permite encontrarlo en la consola del broker. Y mantén el MDB ligero (parsing más una llamada a un service bean), para que la lógica se pueda probar con un test JUnit normal y el MDB una sola vez con un broker embebido.

Qué cambia y qué no en Jakarta Messaging 3.1

Messaging 3.0 fue el cambio de namespace (javax.jms a jakarta.jms); la 3.1, parte de Jakarta EE 10, subió la versión mínima de Java e hizo pequeñas aclaraciones, pero no añadió API nueva. La siguiente generación, discutida desde hace tiempo (listeners nativos de CDI en beans normales, una API reactiva), no ha llegado, así que el MDB sigue siendo la única forma estándar y portable de consumir mensajes en un servidor Jakarta EE. Los fabricantes de servidores cubren el hueco: WildFly y Open Liberty exponen funcionalidades de Artemis y de su propio motor mediante propiedades de activación, y MicroProfile Reactive Messaging ofrece una alternativa basada en anotaciones en los runtimes que lo soportan. Cuando leas código que usa javax.jms, @MessageDriven de javax.ejb y destinationLookup, es el mismo modelo; solo ha cambiado el prefijo.

Errores comunes

• Lanzar excepciones ante fallos permanentes. El mensaje entra en un bucle de reentregas hasta el límite del broker. Envíalo a una cola de errores y retorna con normalidad.
• Procesamiento no idempotente. Una reentrega tras una caída entre el commit y el acuse de recibo, poco frecuente pero real, procesa el mensaje dos veces. Deduplica por id de mensaje o clave de negocio.
• Transacciones gestionadas por el bean en un MDB. El acuse de recibo y tu transacción quedan desacoplados; puedes perder mensajes o procesarlos dos veces. Usa transacciones gestionadas por el contenedor.
• Selectores en colas sin un consumidor que recoja el resto. Los mensajes que no coinciden se acumulan en silencio. Monitoriza la profundidad de la cola o usa un topic.
• Dar por hecho un procesamiento ordenado. El pool procesa en paralelo. Diseña pensando en ello o limita el pool.
• ObjectMessage con clases de dominio. Acoplamiento de versiones y riesgo de deserialización. Usa JSON en un TextMessage.
• Cerrar un JMSContext inyectado. Pertenece al contenedor; cerrarlo lanza una excepción o rompe las llamadas posteriores. Crea tu propio contexto a partir de una ConnectionFactory solo en código que se ejecute fuera de la gestión del contenedor.

Puntos clave

• Un message-driven bean convierte una cola o un topic en una llamada a método: el contenedor gestiona la conexión, un pool de instancias y la transacción alrededor de cada onMessage.
• Las propiedades de activación (destinationLookup, destinationType, messageSelector, ajustes de suscripción) son estándar; el tamaño del pool, los límites de reentrega y el tratamiento de dead letters son configuración del broker y del servidor.
• Con transacciones gestionadas por el contenedor, el acuse de recibo forma parte del commit: un retorno normal confirma y una excepción revierte y provoca la reentrega.
• Inyecta JMSContext para enviar; se une a la transacción JTA en curso, de modo que las escrituras en base de datos y los mensajes salientes se confirman juntos.
• Las suscripciones durables conservan los mensajes de un topic mientras el consumidor está caído, las compartidas permiten que un pool o un clúster los consuma, y los selectores filtran en el broker.
• Los consumidores deben ser idempotentes, validar pronto y desviar los mensajes imposibles de procesar en lugar de lanzar excepciones indefinidamente.
• Messaging 3.1 no cambió nada funcionalmente; el modelo de MDB es el mismo que en Java EE 7, con paquetes jakarta.*.