RabbitMQ avec Spring AMQP et Spring Boot : RabbitTemplate, @RabbitListener et retries

Le client Java de RabbitMQ vous donne des channels, des tableaux d'octets et des callbacks. Spring AMQP vous donne un RabbitTemplate qui publie des objets, une annotation @RabbitListener qui transforme une méthode en consommateur capable de monter en charge, et un RabbitAdmin qui crée les exchanges et les files décrits par vos beans. Spring Boot ajoute par-dessus l'auto-configuration et les propriétés. Résultat : un service dont le code de messagerie tient en quelques dizaines de lignes. Ce guide présente ces lignes, accompagnées des réglages qui les rendent sûres sous charge. Il vise Spring Boot 3.x avec Spring AMQP 3.x ; les principes restent identiques avec Spring AMQP 4, même si quelques noms de classes changent.

Mise en place et connexion

Un seul starter apporte Spring AMQP et le client RabbitMQ :

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

Boot crée une CachingConnectionFactory — une connexion unique dont les channels sont mis en pool et réutilisés entre threads, ce qui est le bon modèle avec AMQP — et la configure à partir des propriétés :

spring:
  rabbitmq:
    addresses: amqps://rabbit-1.internal:5671,amqps://rabbit-2.internal:5671   # cluster nodes, TLS
    username: orders
    password: ${RABBITMQ_PASSWORD}
    virtual-host: shop
    connection-timeout: 5s
    publisher-confirm-type: correlated     # enable confirms with a CorrelationData callback
    publisher-returns: true                # get unroutable messages back
    template:
      mandatory: true                      # required for returns to fire
      retry:
        enabled: true                      # retry a failed publish (broker unavailable)
        max-attempts: 3
        initial-interval: 500ms
    listener:
      simple:
        acknowledge-mode: auto             # ack on success, nack on exception
        prefetch: 20
        concurrency: 2
        max-concurrency: 8
        default-requeue-rejected: false    # rejected messages go to the DLX, not back to the queue

La liste addresses fournit au client plusieurs nœuds à essayer ; la factory avec cache se reconnecte au suivant quand un nœud disparaît, et les consommateurs se réabonnent automatiquement. Tout le reste de ce fichier est expliqué au fil du guide.

Déclarer la topologie sous forme de beans

Spring AMQP représente exchanges, files et bindings par des beans. À la première connexion, RabbitAdmin (auto-configuré par Boot) déclare tous les Declarable du contexte et les redéclare après une reconnexion. Ces déclarations sont idempotentes : plusieurs instances du service peuvent démarrer dans n'importe quel ordre.

@Configuration
class OrdersTopology {

  static final String EXCHANGE = "orders";
  static final String QUEUE = "fulfilment";
  static final String DLX = "orders.dlx";

  @Bean TopicExchange ordersExchange() { return ExchangeBuilder.topicExchange(EXCHANGE).durable(true).build(); }

  @Bean DirectExchange ordersDlx() { return ExchangeBuilder.directExchange(DLX).durable(true).build(); }

  @Bean Queue fulfilmentQueue() {
    return QueueBuilder.durable(QUEUE)
        .quorum()                              // x-queue-type=quorum
        .deadLetterExchange(DLX)
        .deadLetterRoutingKey(QUEUE)
        .deliveryLimit(5)                      // quorum queues: dead-letter after 5 redeliveries
        .build();
  }

  @Bean Queue fulfilmentDlq() { return QueueBuilder.durable(QUEUE + ".dlq").quorum().build(); }

  @Bean Binding fulfilmentBinding() {
    return BindingBuilder.bind(fulfilmentQueue()).to(ordersExchange()).with("order.*.placed");
  }

  @Bean Binding fulfilmentDlqBinding() {
    return BindingBuilder.bind(fulfilmentDlq()).to(ordersDlx()).with(QUEUE);
  }
}

Deux habitudes gardent cet ensemble lisible à mesure que le service grandit. Placez les noms dans des constantes (ou dans une classe @ConfigurationProperties) pour que producteurs, listeners et tests s'appuient tous sur les mêmes valeurs, et regroupez les déclarations liées dans un bean Declarables quand elles sont nombreuses, plutôt qu'un bean par binding. N'oubliez pas que les arguments d'une file sont immuables : si vous modifiez plus tard deliveryLimit sur une file qui existe déjà, RabbitMQ refuse la déclaration avec une erreur PRECONDITION_FAILED et le listener ne démarre pas. Supprimez la file ou exprimez le réglage sous forme de policy.

Publier avec RabbitTemplate

RabbitTemplate.convertAndSend sérialise un objet avec le MessageConverter configuré, positionne le content type et publie le message sur un exchange avec une routing key. Par défaut, le convertisseur est SimpleMessageConverter, qui gère les chaînes, les tableaux d'octets et la sérialisation Java ; déclarez un convertisseur JSON et Boot l'injecte à la fois dans le template et dans la factory des listener containers :

@Bean
MessageConverter jsonMessageConverter() {
  return new Jackson2JsonMessageConverter();   // Spring AMQP 3.x; JacksonJsonMessageConverter in 4.x
}

@Service
class OrderEventPublisher {
  private final RabbitTemplate rabbit;

  OrderEventPublisher(RabbitTemplate rabbit) {
    this.rabbit = rabbit;
    rabbit.setConfirmCallback((correlation, ack, cause) -> {
      if (!ack) log.error("broker rejected {}: {}", correlation != null ? correlation.getId() : "?", cause);
    });
    rabbit.setReturnsCallback(returned ->
      log.error("unroutable message to {} with key {}: {}",
          returned.getExchange(), returned.getRoutingKey(), returned.getReplyText()));
  }

  public void orderPlaced(OrderPlaced event) {
    rabbit.convertAndSend(OrdersTopology.EXCHANGE, "order." + event.region() + ".placed", event,
        message -> {
          message.getMessageProperties().setMessageId(event.orderId());   // for consumer-side de-duplication
          message.getMessageProperties().setHeader("schemaVersion", 2);
          return message;
        },
        new CorrelationData(event.orderId()));
  }
}

Avec publisher-confirm-type: correlated, le broker confirme ou rejette chaque publication et le template appelle votre ConfirmCallback avec le CorrelationData que vous avez fourni. Un nack signifie que le broker n'a pas pu prendre la responsabilité du message — un nœud perdu en pleine publication, ou une quorum queue privée de majorité — et la bonne réaction consiste à réessayer ou à consigner l'échec, pas à écrire un log puis à passer à autre chose. Le returns callback se déclenche quand mandatory vaut true et que la routing key ne correspond à aucun binding : c'est le seul moyen de repérer une faute de frappe dans une routing key.

C'est dans la lambda MessagePostProcessor que l'on positionne les en-têtes, les identifiants de message, les priorités et les expirations. Le mode de livraison persistant est la valeur par défaut du template.

Consommer avec @RabbitListener

Une méthode @RabbitListener devient un consommateur adossé à un listener container. Boot crée un SimpleMessageListenerContainer par annotation à partir des propriétés spring.rabbitmq.listener.simple.* ; le payload est converti par le même convertisseur JSON, et les en-têtes peuvent être injectés.

@Component
class FulfilmentListener {

  @RabbitListener(queues = OrdersTopology.QUEUE, concurrency = "2-8")
  public void onOrderPlaced(OrderPlaced event,
                            @Header(AmqpHeaders.MESSAGE_ID) String messageId,
                            @Header(name = "x-delivery-count", required = false) Long deliveryCount) {
    if (fulfilments.alreadyProcessed(messageId)) return;          // idempotent: duplicates are harmless
    fulfilments.fulfil(event);                                    // throwing here → nack → DLX (requeue disabled)
  }
}

concurrency = "2-8" démarre deux threads consommateurs et monte jusqu'à huit tant que la file accumule du retard. Chaque thread possède son propre channel avec le prefetch configuré : huit consommateurs et un prefetch de 20 peuvent donc avoir 160 messages non acquittés en même temps. Dimensionnez le prefetch en gardant cette multiplication en tête.

Avec acknowledge-mode: auto (la valeur par défaut), le container acquitte après le retour de la méthode et envoie un acquittement négatif quand elle lève une exception. Comme default-requeue-rejected vaut false dans la configuration ci-dessus, un message rejeté part vers le dead-letter exchange au lieu de revenir en tête de file. Pour rejeter volontairement un message quel que soit ce réglage, levez AmqpRejectAndDontRequeueException ; pour demander une remise en file quand un système en aval est brièvement indisponible, levez ImmediateRequeueAmqpException.

Acquittement manuel

Certaines charges de travail doivent acquitter à un moment précis — après un commit en base de données, ou par lots. Passez le container en AcknowledgeMode.MANUAL et injectez le channel et le delivery tag :

@RabbitListener(queues = OrdersTopology.QUEUE, ackMode = "MANUAL")
public void onOrderPlaced(OrderPlaced event, Channel channel,
                          @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
  try {
    fulfilments.fulfil(event);
    channel.basicAck(tag, false);
  } catch (DownstreamUnavailable e) {
    channel.basicNack(tag, false, true);      // requeue: another instance may succeed
  } catch (RuntimeException e) {
    channel.basicNack(tag, false, false);     // dead-letter it
  }
}

En mode manuel, le container n'acquitte pas à votre place : une méthode qui oublie l'ack laisse le message non acquitté jusqu'à la fermeture du channel — et le consumer timeout du broker le ferme au bout de trente minutes, avec une erreur.

Déclarer la file sur le listener

Pour les petits services, il est pratique de déclarer le binding directement dans l'annotation ; RabbitAdmin le prend en compte comme n'importe quel autre declarable :

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "notifications", durable = "true",
                   arguments = @Argument(name = "x-queue-type", value = "quorum")),
    exchange = @Exchange(value = "orders", type = ExchangeTypes.TOPIC),
    key = "order.#"))
public void notify(OrderPlaced event) { ... }

Retries et dead letters

Un message en échec peut être relancé à trois niveaux, et un service robuste en combine deux.

Le retry au niveau du container se résume à un changement de propriétés :

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true
          max-attempts: 4          # the original delivery plus three retries
          initial-interval: 1s
          multiplier: 2.0          # 1 s, 2 s, 4 s
          max-interval: 10s
        default-requeue-rejected: false

Une fois les retries épuisés, le MessageRecoverer du container décide du sort du message. Le RejectAndDontRequeueRecoverer par défaut le rejette sans remise en file, ce qui l'envoie vers le dead-letter exchange configuré sur la file. RepublishMessageRecoverer est une alternative qui publie le message sur l'exchange de votre choix avec des en-têtes décrivant l'exception et la stack trace — bien plus utile quand quelqu'un ouvre plus tard la dead-letter queue pour comprendre ce qui s'est passé :

@Bean
MessageRecoverer messageRecoverer(RabbitTemplate template) {
  return new RepublishMessageRecoverer(template, OrdersTopology.DLX, OrdersTopology.QUEUE);
}

Quel que soit le mécanisme, une dead-letter queue a besoin d'un responsable : une alerte sur sa profondeur, un listener qui consigne l'échec dans un endroit visible et une procédure documentée pour rejouer les messages une fois le bug corrigé — en général la fonction move messages de l'interface de management (avec le plugin shovel) ou un petit endpoint d'administration qui lit la dead-letter queue et republie.

Request-reply

RabbitTemplate.convertSendAndReceive implémente l'appel de procédure distante sur AMQP : il publie une requête, attend une réponse portant le même identifiant de corrélation et convertit le résultat. Par défaut, il utilise la pseudo-file direct reply-to de RabbitMQ (amq.rabbitmq.reply-to), ce qui évite de créer une file temporaire à chaque appel et reste assez rapide pour un usage interactif.

PriceQuote quote = rabbit.convertSendAndReceiveAsType(
    "pricing", "quote.request", new QuoteRequest("A-100", 3),
    new ParameterizedTypeReference<PriceQuote>() {});

Côté serveur, une méthode @RabbitListener se contente de retourner une valeur ; le container l'envoie vers le replyTo de la requête avec le même correlationId :

@RabbitListener(queues = "pricing.requests")
public PriceQuote quote(QuoteRequest request) {
  return pricing.quote(request.sku(), request.quantity());
}

Réglez spring.rabbitmq.template.reply-timeout (cinq secondes par défaut) sur une valeur cohérente avec le timeout de l'appelant, et gardez à l'esprit que le request-reply réintroduit le couplage temporel que la messagerie devait supprimer : si le service de tarification est en panne, l'appelant échoue après le timeout, exactement comme avec un appel HTTP.

Tester contre un vrai broker

Mocker RabbitTemplate ne teste rien des routing keys, des convertisseurs ni des acquittements. Testcontainers démarre un vrai RabbitMQ dans Docker, et le @ServiceConnection de Boot (3.1+) renseigne les propriétés de connexion sans aucune configuration manuelle :

@SpringBootTest
@Testcontainers
class FulfilmentFlowTest {

  @Container @ServiceConnection
  static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:4.0-management");

  @Autowired OrderEventPublisher publisher;
  @Autowired FulfilmentRepository fulfilments;

  @Test
  void placedOrderIsFulfilled() {
    publisher.orderPlaced(new OrderPlaced("o-1", "eu", 149.90));
    Awaitility.await().atMost(Duration.ofSeconds(5))
        .untilAsserted(() -> assertThat(fulfilments.find("o-1")).isPresent());
  }
}

Comme les beans de topologie déclarent tout au démarrage, le test tourne contre un broker vide, sans fixtures. Prenez le tag d'image -management si vous voulez ouvrir l'interface sur le port mappé pendant le débogage d'un test qui échoue.

Erreurs fréquentes

• Oublier le convertisseur JSON. Les listeners reçoivent des byte[] ou échouent sur une ClassCastException ; les producteurs envoient des payloads en sérialisation Java que les consommateurs non Java ne savent pas lire. Déclarez le bean de conversion une fois pour toutes.
• Laisser default-requeue-rejected à true. Un listener qui lève une exception sur un message mal formé tourne à plein régime sur ce même message, indéfiniment. Passez-le à false et envoyez en dead letter.
• Ignorer les callbacks de confirm et de return. Les publications disparaissent sans bruit au moindre hoquet du broker ou à la moindre faute dans une routing key. Activez les confirms correlated et mandatory, puis journalisez ou réessayez dans les callbacks.
• Modifier les arguments d'une file existante. PRECONDITION_FAILED au démarrage. Supprimez la file hors production ; utilisez des policies en production.
• Bloquer sur des retries à longs intervalles. Un thread de listener qui dort plusieurs minutes entre deux tentatives est un thread que vous n'avez plus. Gardez des retries courts dans le container et reportez les longues attentes sur une file à TTL.
• Partager un état mutable entre threads de listener concurrents. concurrency = "2-8" signifie que votre bean listener est appelé depuis huit threads à la fois ; gardez-le sans état ou synchronisez correctement.

Points clés à retenir

• Le spring-boot-starter-amqp de Boot fournit une connection factory avec cache, un RabbitTemplate, un RabbitAdmin et des listener containers configurés par les propriétés spring.rabbitmq.*.
• Déclarez exchanges, files et bindings sous forme de beans ; ils sont créés de manière idempotente à la première connexion, et la quorum queue associée à un dead-letter exchange devrait être la forme par défaut.
• Publiez avec un MessageConverter JSON, des confirms corrélés et des returns mandatory, pour savoir ce que le broker a réellement accepté.
• @RabbitListener monte en charge avec concurrency, est borné par le prefetch, acquitte automatiquement au retour de la méthode et rejette en cas d'exception ; réservez le mode manuel aux cas où le point d'acquittement doit être précis.
• Faites des retries courts dans le container, puis passez en dead letter via le broker ; vérifiez que default-requeue-rejected vaut false et que la dead-letter queue a un responsable.
• Testez le vrai flux avec Testcontainers et @ServiceConnection ; les mocks ne détectent ni les erreurs de routage ni les erreurs de sérialisation.