L'API JMS (Jakarta Messaging 3.1) : sessions, messages, acquittement et sélecteurs

Jakarta Messaging — que tout le monde appelle encore JMS — est l'API qu'utilise une application Java pour envoyer des messages à un broker et en recevoir. Cette API est un contrat, pas un produit : ActiveMQ Classic, ActiveMQ Artemis, IBM MQ, Oracle AQ et bien d'autres en fournissent une implémentation, et un code qui s'en tient aux interfaces jakarta.jms passe de l'une à l'autre par simple changement de configuration. Ce guide cible JMS 3.1 (Jakarta EE 10) et s'appuie sur ActiveMQ Classic 6 comme broker pour les exemples, parce qu'il est léger et gratuit à faire tourner en local.

Deux API, un seul modèle

Depuis la version 2.0, JMS propose deux styles de programmation.

L'API classique (JMS 1.1) construit une chaîne d'objets : une ConnectionFactory crée une Connection, la connexion crée une Session, et la session crée les objets MessageProducer, MessageConsumer et Message. Elle est verbeuse mais explicite, et c'est elle que les serveurs d'applications utilisent en coulisses pour les message-driven beans.

L'API simplifiée (à partir de JMS 2.0) fusionne connexion et session dans un unique JMSContext et lui adjoint des helpers fluides, JMSProducer et JMSConsumer. Tout ce que permet l'API classique, l'API simplifiée le fait en moins de lignes et avec des ressources AutoCloseable. C'est elle qu'il faut utiliser pour du nouveau code.

Les deux styles reposent sur les mêmes objets, et toutes les notions qui suivent — destinations, types de messages, acquittement, sélecteurs — s'appliquent à l'un comme à l'autre.

Ajoutez au projet l'API et un client de broker. L'artefact de l'API fournit les interfaces ; l'artefact client fournit l'implémentation :

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

Envoyer un message

La ConnectionFactory est le seul objet propre à l'éditeur que vous manipulez. Dans une application autonome, vous l'instanciez vous-même ; dans un serveur d'applications, vous la récupérez via JNDI ou l'injectez avec @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}");
    }
  }
}

Sans argument, createContext() fournit le mode AUTO_ACKNOWLEDGE ; createContext(JMSContext.SESSION_TRANSACTED) fournit un contexte transactionnel sur lequel vous devez appeler commit(). createQueue("orders.placed") ne crée rien sur le broker : il construit côté client l'identité d'une destination. ActiveMQ Classic crée la file à la première utilisation ; d'autres brokers exigent qu'un administrateur l'ait définie au préalable, ce qui explique que la spécification traite les destinations comme des objets administrés, normalement obtenus via JNDI.

Le JMSProducer renvoyé par createProducer() est léger et réutilisable : définissez-y le mode de livraison, la priorité, la durée de vie et les propriétés, puis appelez send autant de fois que nécessaire.

Recevoir un message

Les consommateurs prennent deux formes : des appels bloquants à receive, ou des callbacks asynchrones 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 forme asynchrone enregistre un listener ; la bibliothèque cliente appelle onMessage sur son propre thread à chaque livraison :

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

Deux règles comptent ici. Une session (et donc un JMSContext) est mono-thread : ne la partagez jamais entre threads et ne l'appelez pas depuis un listener qui lui appartient. Par ailleurs, la connexion doit être démarrée pour que les consommateurs reçoivent quoi que ce soit — JMSContext s'en charge automatiquement, alors que l'API classique exige un connection.start() explicite, ce qui est de très loin la première cause du fameux « mon consommateur ne reçoit rien ».

Types de messages, en-têtes et propriétés

Un message JMS se compose d'en-têtes, de propriétés et d'un corps typé. Choisissez le type de corps en fonction de ce que le consommateur doit en faire.

Les en-têtes sont des champs fixes définis par la spécification. Le producteur en renseigne certains (JMSDeliveryMode, JMSPriority, JMSExpiration via la durée de vie, JMSReplyTo, JMSCorrelationID, JMSType et, depuis JMS 2.0, JMSDeliveryTime pour la livraison différée) ; le fournisseur renseigne les autres (JMSMessageID, JMSTimestamp, JMSDestination, JMSRedelivered). La priorité va de 0 à 9, avec 4 par défaut, et les brokers sont seulement tenus d'essayer de livrer d'abord les priorités les plus hautes.

Les propriétés sont vos propres paires clé-valeur, de type booléen, numérique ou String. Les propriétés dont le nom commence par JMSX sont réservées : JMSXGroupID et JMSXGroupSeq regroupent des messages liés pour qu'un seul consommateur voie toute une séquence dans l'ordre, et JMSXDeliveryCount (obligatoire depuis la 2.0) indique au consommateur combien de fois il a déjà vu ce message — la base du traitement des messages empoisonnés.

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

Le mode de livraison mérite qu'on s'y arrête. DeliveryMode.PERSISTENT (la valeur par défaut) demande au broker d'écrire le message dans son stockage avant d'acquitter l'envoi, si bien que le message survit à un redémarrage du broker ; le producteur reste bloqué jusqu'à réception de cet acquittement. NON_PERSISTENT fait l'économie du disque et de l'attente : c'est plusieurs fois plus rapide, et adapté aux métriques, aux heartbeats et à tout ce que vous pouvez vous permettre de perdre.

Modes d'acquittement : ce qui est réellement promis

Un message reste dans la file tant que le consommateur ne l'a pas acquitté. La manière dont se fait cet acquittement, c'est le mode d'acquittement de la session, et c'est lui qui détermine votre garantie de livraison.

Deux subtilités piègent régulièrement les développeurs. D'abord, CLIENT_ACKNOWLEDGE porte sur toute la session : acquitter le message 5 acquitte aussi les messages 1 à 4, ce n'est donc pas un outil message par message (ActiveMQ propose pour cela un INDIVIDUAL_ACKNOWLEDGE non standard). Ensuite, une session transactionnelle regroupe les envois et les réceptions : consommer depuis orders.placed et envoyer vers invoices.created dans une même transaction signifie que les deux ont lieu ou qu'aucun n'a lieu, du point de vue du broker. Votre base de données n'en fait pas partie — il faut pour cela un gestionnaire de transactions XA via un serveur d'applications, ou le pattern 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
  }
}

En cas de relivraison, le broker positionne JMSRedelivered et incrémente JMSXDeliveryCount. Le client ActiveMQ applique une politique de relivraison (par défaut, six tentatives espacées d'une seconde), puis envoie le message dans une dead-letter queue nommée ActiveMQ.DLQ ; allez voir cette file chaque fois que des messages « disparaissent ».

Sélecteurs

Un sélecteur de messages permet à un consommateur de ne recevoir que les messages dont les en-têtes ou les propriétés satisfont une expression. Sa syntaxe est un sous-ensemble des clauses WHERE de SQL-92 ; c'est le broker qui l'évalue, de sorte que les messages qui ne correspondent pas ne transitent jamais sur le réseau jusqu'à ce consommateur.

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

Les opérateurs pris en charge comprennent les comparaisons, AND/OR/NOT, BETWEEN, IN, LIKE avec les jokers % et _, IS NULL et l'arithmétique. Les identifiants désignent des noms de propriétés ou les en-têtes préfixés par JMS qui sont sélectionnables (JMSDeliveryMode, JMSPriority, JMSMessageID, JMSTimestamp, JMSCorrelationID, JMSType). Sur une file, les sélecteurs ont un coût : une file dont les seuls consommateurs utilisent des sélecteurs qui ne correspondent à rien accumule indéfiniment les messages non retenus. Sur un topic, ils sont gratuits, car les messages qui ne correspondent pas ne sont tout simplement pas livrés à cet abonnement.

Topics : abonnements durables et partagés

Un consommateur de topic ordinaire reçoit les messages publiés pendant qu'il est connecté, et rien d'autre. Deux types d'abonnements changent la donne.

Un abonnement durable est identifié par un nom (et, dans l'API classique, par un identifiant client). Le broker conserve les messages qui lui sont destinés pendant que l'abonné est hors ligne et lui livre l'arriéré à la reconnexion :

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

Jusqu'à JMS 2.0, un abonnement durable n'admettait qu'un seul consommateur actif, ce qui rendait impossible la montée en charge d'un consommateur de topic sans astuces propres au fournisseur, comme les virtual topics d'ActiveMQ. Les abonnements partagés corrigent cela : plusieurs consommateurs utilisant le même nom d'abonnement partagé se répartissent les messages, à la manière d'une file.

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

Utilisez createSharedConsumer lorsque l'abonnement n'a pas besoin de survivre à une déconnexion, et createSharedDurableConsumer lorsqu'il le doit. Pensez à appeler unsubscribe(name) quand un abonnement durable est mis hors service ; sinon le broker stocke des messages pour un abonné qui ne reviendra jamais, et le disque finit par se remplir.

Request-reply sur des files

JMS n'offre aucune primitive d'appel distant, mais les en-têtes permettent d'en construire une robuste sans effort. Le demandeur crée une file temporaire (une destination qui n'existe que le temps de sa connexion), envoie la requête avec JMSReplyTo pointant vers elle et un JMSCorrelationID tout neuf, puis attend une réponse dont l'identifiant de corrélation correspond.

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

Le service qui répond lit getJMSReplyTo() et getJMSCorrelationID() sur la requête et envoie sa réponse vers cette destination avec le même identifiant de corrélation. Donnez un timeout à chaque requête : un demandeur qui attend sans timeout, c'est une fuite de threads qui n'attend qu'un hoquet du broker.

JMS dans Spring : JmsTemplate et @JmsListener

La plupart des équipes Java n'appellent jamais l'API brute : le JmsTemplate de Spring gère la connexion, la session et la libération des ressources, et @JmsListener transforme n'importe quelle méthode en consommateur adossé à un pool de threads d'écoute. À partir de Spring Boot 3.2, spring-boot-starter-activemq auto-configure une ConnectionFactory à partir de spring.activemq.broker-url et l'enveloppe dans une factory avec cache, pour que chaque send n'ouvre pas une nouvelle connexion 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
  }
}

Positionnez spring.jms.listener.session.transacted=true (la valeur par défaut de Boot pour le conteneur de listeners) pour qu'une exception dans une méthode d'écoute annule la réception du message au lieu de le perdre, et ne positionnez spring.jms.pub-sub-domain=true que si les noms de destinations désignent des topics.

Erreurs courantes

• Oublier connection.start() avec l'API classique. Le consommateur se connecte, la file contient des messages, et rien n'arrive. JMSContext démarre la connexion de manière paresseuse, à la création du premier consommateur, ce qui est une raison de plus de le préférer.
• Créer une connexion par message. Les connexions sont des sessions TCP coûteuses, avec leurs propres threads. Créez une seule ConnectionFactory et un petit nombre de connexions à longue durée de vie ; dans Spring, utilisez CachingConnectionFactory ou la configuration par défaut de Boot.
• Partager une session entre threads. Les sessions ne sont pas thread-safe. Chaque thread de travail dispose de sa propre session (ou de son propre JMSContext) ; partagez la connexion, pas la session.
• Utiliser ObjectMessage. Il oblige les deux côtés à disposer de la même classe dans le classpath et ouvre la porte aux vulnérabilités de désérialisation Java. Envoyez du JSON dans un TextMessage ; utilisez le formateur JSON pour inspecter les payloads pendant le débogage.
• Des sélecteurs qui ne correspondent jamais sur une file. Les messages s'accumulent. Routez plutôt vers des files différentes dès l'envoi, ou assurez-vous qu'un consommateur accepte tout.
• Ignorer la dead-letter queue. Les messages empoisonnés finissent silencieusement dans ActiveMQ.DLQ (ou son équivalent chez votre fournisseur). Surveillez-la et prévoyez un moyen d'en inspecter et d'en rejouer le contenu.

Points clés à retenir

• JMS est une API ; c'est le client de broker ajouté au classpath qui détermine le protocole réseau utilisé et le broker que vous pouvez joindre.
• Préférez JMSContext et le try-with-resources ; gardez des sessions mono-thread et des connexions à longue durée de vie.
• Envoyez du JSON dans un TextMessage, placez les métadonnées de routage dans les propriétés, et utilisez JMSXGroupID quand un ensemble de messages doit rester ordonné.
• Le mode d'acquittement définit votre garantie : AUTO et CLIENT offrent l'at-least-once, DUPS_OK échange des doublons contre de la vitesse, et les sessions transactionnelles rendent atomique le couple consommation-production côté broker.
• Les sélecteurs ne filtrent que sur les en-têtes et les propriétés ; les abonnements durables partagés permettent de faire monter en charge les consommateurs de topics comme sur une file.
• Un request-reply, c'est une file temporaire plus JMSReplyTo plus JMSCorrelationID, toujours avec un timeout.
• Dans Spring, JmsTemplate et @JmsListener éliminent le code répétitif sans rien changer aux garanties ci-dessus.