RabbitMQ et AMQP 0-9-1 : exchanges, bindings, files et routage

RabbitMQ est le broker qui implémente le plus fidèlement AMQP 0-9-1, et presque tout ce qui déroute les débutants vient d'une méconnaissance du modèle de ce protocole. Un producteur n'envoie jamais dans une file. Il envoie à un exchange, l'exchange consulte ses bindings, et des copies du message atterrissent dans zéro, une ou plusieurs files, où les consommateurs viennent les chercher. Une fois ce schéma bien en tête, la flexibilité de RabbitMQ — un même message envoyé à une file de travail, à un journal d'audit et à un flux de tableau de bord sans que le producteur le sache — devient un outil de conception plutôt qu'une source de frustration. Ce guide vise RabbitMQ 3.13 et 4.x avec le client Java officiel.

Le modèle AMQP 0-9-1

Cinq notions suffisent à décrire le protocole.

Une connexion est une connexion TCP longue durée, généralement chiffrée en TLS, authentifiée auprès d'un virtual host (vhost) — un espace de noms regroupant exchanges, files et permissions, qui permet à un même broker de servir plusieurs applications de façon isolée. À l'intérieur d'une connexion, les applications ouvrent des channels : des sessions multiplexées peu coûteuses sur lesquelles se font toutes les publications et consommations. La règle est une connexion par processus et un channel par thread, car les channels ne sont pas thread-safe et les connexions coûtent cher.

Un exchange reçoit les messages des producteurs et les achemine. Il ne stocke rien. Une file (queue) stocke les messages et les livre aux consommateurs. Un binding relie un exchange à une file et peut porter une binding key (ou, pour les exchanges headers, un ensemble d'en-têtes). Lorsqu'un producteur envoie un message, il désigne un exchange et fournit une routing key ; c'est le type d'exchange qui détermine comment la routing key est comparée aux binding keys.

Tout cela est programmable : les applications déclarent exchanges, files et bindings via le protocole, et les déclarations sont idempotentes tant que les arguments sont identiques. C'est pourquoi un producteur et un consommateur peuvent tous deux déclarer la file qu'ils partagent et démarrer dans n'importe quel ordre.

publisher --(exchange, routing key)--> exchange --bindings--> queue(s) --> consumer(s)

Les quatre types d'exchange

Direct

Un exchange direct achemine un message vers chaque file dont la binding key est strictement égale à la routing key. C'est l'outil de la livraison point à point et de la répartition des messages dans un petit ensemble fixe de files, par exemple une file par niveau de sévérité des logs.

L'exchange par défaut est un exchange direct prédéclaré dont le nom est la chaîne vide. Chaque file y est automatiquement liée avec son propre nom comme binding key, si bien que basicPublish("", "orders", ...) livre le message à la file nommée orders. C'est pratique pour les tutoriels et les files de réponse, et un piège pour tout le reste : les producteurs finissent par coder en dur des noms de files, soit précisément le couplage que les exchanges sont censés supprimer.

Fanout

Un exchange fanout ignore la routing key et copie le message vers toutes les files liées. C'est la primitive de diffusion : liez une file par abonné, et chacun reçoit sa propre copie, à son rythme, avec ses propres acquittements.

Topic

Un exchange topic compare la routing key à des motifs de binding. Les routing keys sont des mots séparés par des points, jusqu'à 255 octets ; dans une binding key, * correspond à exactement un mot et # à zéro mot ou plus.

Un exchange topic avec une binding key # se comporte comme un fanout, et avec une clé sans joker comme un exchange direct. C'est pourquoi de nombreuses équipes utilisent un seul exchange topic par bounded context et expriment tout sous forme de conventions de routing key, comme <entity>.<region>.<event>.

Headers

Un exchange headers ignore la routing key et compare les en-têtes du message aux arguments du binding ; l'argument x-match à all exige que tous les en-têtes listés correspondent, et any qu'au moins un corresponde. Il route sur plusieurs attributs à la fois, au prix d'être plus lent et moins lisible que les clés topic. La plupart des conceptions qui y font appel gagneraient à encoder ces attributs dans une routing key.

Déclarer la topologie en Java

Le client officiel est com.rabbitmq:amqp-client (5.x). Ce producteur déclare un exchange topic durable et une quorum queue durable, les lie, puis publie un message persistant :

import com.rabbitmq.client.*;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class OrderPublisher {
  public static void main(String[] args) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUri("amqps://orders:[email protected]:5671/shop");  // vhost "shop"

    try (Connection connection = factory.newConnection("order-service");
         Channel channel = connection.createChannel()) {

      channel.exchangeDeclare("orders", BuiltinExchangeType.TOPIC, /* durable */ true);
      channel.queueDeclare("fulfilment", /* durable */ true, /* exclusive */ false,
          /* autoDelete */ false, Map.of("x-queue-type", "quorum"));
      channel.queueBind("fulfilment", "orders", "order.*.placed");

      channel.confirmSelect();                      // enable publisher confirms on this channel

      AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
          .contentType("application/json")
          .deliveryMode(2)                          // 2 = persistent: written to disk by durable queues
          .messageId("o-9911")
          .build();

      channel.basicPublish("orders", "order.eu.placed", /* mandatory */ true, props,
          "{\"orderId\":\"o-9911\",\"total\":149.90}".getBytes(StandardCharsets.UTF_8));

      channel.waitForConfirmsOrDie(5_000);          // throws if the broker did not confirm in 5 s
    }
  }
}

Trois flags de ce code décident de la durabilité, et les trois sont nécessaires pour qu'un message survive au redémarrage du broker : l'exchange doit être durable (sinon il disparaît avec ses bindings), la file doit être durable, et le message doit être publié en delivery mode 2. Les quorum queues sont toujours durables et persistent toujours leurs messages ; le flag delivery mode reste important pour les files classiques.

Les types de files

RabbitMQ propose trois sortes de files, choisies avec l'argument x-queue-type au moment de la déclaration et immuables ensuite.

Une quorum queue conserve ses messages sur une majorité des nœuds du cluster et élit un leader via Raft ; elle survit à la perte d'une minorité de nœuds sans perdre de messages acquittés, et c'est cette propriété qui en a fait le choix recommandé par défaut. Elle compte aussi le nombre de relivraisons d'un message dans l'en-tête x-delivery-count et peut basculer les messages vers un dead-letter exchange au-delà d'une limite x-delivery-limit configurable, ce que les files classiques ne savent pas faire. En contrepartie, elle consomme plus de disque et de mémoire par message et affiche une latence plus élevée qu'une file classique mono-nœud.

Un stream est un journal en ajout seul : les consommateurs lisent à partir d'un offset, plusieurs consommateurs peuvent lire les mêmes messages, et la rétention se règle par taille ou par âge plutôt que par consommation. Les streams sont accessibles en AMQP 0-9-1 via le client habituel, mais le protocole et le client stream dédiés sont bien plus rapides. Si vous avez envie d'un rejeu à la Kafka dans RabbitMQ, les streams sont la réponse.

RabbitMQ 4.0 a aussi fait d'AMQP 1.0 un protocole de premier plan aux côtés de 0-9-1 : une application Java peut donc utiliser le client Qpid JMS ou le client AMQP 1.0 de RabbitMQ sur les mêmes files ; le modèle exchanges et bindings décrit ici reste inchangé.

Consommer, acquitter et prefetch

Un consommateur s'abonne à une file avec basicConsume, et le broker lui pousse les livraisons. La suite dépend du mode d'acquittement choisi.

Avec autoAck = true, le broker oublie un message dès qu'il l'a écrit sur la socket. Si le consommateur plante, tout ce qui était en vol est perdu. Avec autoAck = false, le message reste dans la file, marqué unacked, jusqu'à ce que le consommateur envoie basicAck ; si le channel se ferme avant, le broker remet le message en file pour un autre consommateur. C'est la livraison at-least-once.

channel.basicQos(20);   // prefetch: at most 20 unacknowledged deliveries per consumer on this channel

channel.basicConsume("fulfilment", /* autoAck */ false, "fulfilment-worker-1",
    new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String tag, Envelope env, AMQP.BasicProperties props, byte[] body) throws IOException {
        long deliveryTag = env.getDeliveryTag();
        try {
          fulfil(new String(body, StandardCharsets.UTF_8));
          channel.basicAck(deliveryTag, /* multiple */ false);
        } catch (TransientFailure e) {
          channel.basicNack(deliveryTag, false, /* requeue */ true);    // try again, possibly elsewhere
        } catch (RuntimeException e) {
          channel.basicNack(deliveryTag, false, /* requeue */ false);   // dead-letter it, if a DLX is set
        }
      }
    });

basicQos fixe le prefetch count : le nombre de livraisons que le broker peut pousser vers un consommateur avant de recevoir des acquittements. Sans lui, le broker envoie aussi vite que la socket le permet, un consommateur accumule des milliers de messages en mémoire pendant que les autres se tournent les pouces, et un plantage les remet tous en file. Un prefetch entre 10 et 100 convient à la plupart des charges ; 1 offre une équité parfaite au prix d'un aller-retour par message.

Deux garde-fous protègent le broker des consommateurs qui se comportent mal. Relivrer avec requeue = true un message qui échoue à chaque fois crée une boucle infinie qui brûle du CPU ; vérifiez env.isRedeliver() ou, sur les quorum queues, l'en-tête x-delivery-count, et cessez de remettre en file après quelques tentatives. Par ailleurs, un consommateur qui garde une livraison sans l'acquitter au-delà du consumer timeout (30 minutes par défaut) voit son channel fermé par le broker, une surprise fréquente avec les consommateurs de type batch.

Publisher confirms

Les acquittements protègent le côté consommateur ; les publisher confirms protègent le producteur. Après confirmSelect, le broker envoie un basic.ack pour chaque message une fois celui-ci routé et, pour les messages persistants dans des files durables, écrit sur disque, ou un basic.nack s'il n'a pas pu le traiter. waitForConfirmsOrDie en est la forme synchrone simple ; pour le débit, publiez par lots ou enregistrez un ConfirmListener qui suit les numéros de séquence en attente, et republiez tout ce qui a été nacké. Combinés à mandatory = true et à un ReturnListener, qui renvoie les messages n'ayant correspondu à aucun binding, les confirms donnent un producteur qui sait exactement ce que le broker a accepté.

TTL, limites et dead-lettering

Les files acceptent des arguments optionnels qui déterminent le sort des messages non consommés.

• x-message-ttl — nombre de millisecondes pendant lesquelles un message peut attendre avant d'être supprimé ou dead-letteré. La propriété expiration fait la même chose pour un message donné.
• x-expires — supprime la file elle-même après ce nombre de millisecondes sans consommateur ni activité ; utile pour les files de réponse propres à chaque client.
• x-max-length / x-max-length-bytes avec x-overflow — plafonne la file ; drop-head supprime les messages les plus anciens, reject-publish fait échouer (nack) les nouvelles publications sous confirms, pour que le producteur perçoive la contre-pression.
• x-dead-letter-exchange et x-dead-letter-routing-key — la destination des messages rejetés, expirés ou en débordement.

Un dead-letter exchange (DLX) est un exchange ordinaire ; en y liant une file, vous obtenez un endroit où les messages en échec s'accumulent, avec un en-tête x-death qui indique pourquoi et combien de fois. La combinaison TTL et DLX est aussi la façon dont RabbitMQ implémente un retry différé sans plugin : nackez le message en échec vers une file « d'attente » dont le TTL est de trente secondes et dont le dead-letter exchange renvoie vers la file de travail.

Map<String, Object> args = Map.of(
    "x-queue-type", "quorum",
    "x-dead-letter-exchange", "orders.dlx",
    "x-delivery-limit", 5,               // quorum queues: dead-letter after 5 redeliveries
    "x-message-ttl", 86_400_000);        // one day
channel.queueDeclare("fulfilment", true, false, false, args);

Les arguments fixés à la déclaration sont figés ; les modifier impose de supprimer la file. Préférez les policies pour tout ce qui peut évoluer : rabbitmqctl set_policy dlx "^fulfilment$" '{"dead-letter-exchange":"orders.dlx"}' --apply-to queues applique le même réglage sans redéclaration, et peut être modifié à chaud.

Administrer le broker

Le plugin de management fournit une interface web et une API HTTP sur le port 15672 (AMQP lui-même écoute sur 5672, TLS sur 5671, les métriques Prometheus sur 15692). Il affiche exchanges, bindings, profondeur des files, débits de messages, connexions et channels, et permet de publier et de récupérer des messages à la main pour déboguer. Le compte par défaut guest/guest ne peut se connecter que depuis localhost ; créez de vrais utilisateurs et accordez-leur des permissions par vhost avant que quoi que ce soit d'autre ne se connecte :

rabbitmqctl add_vhost shop
rabbitmqctl add_user orders 'change-me'
rabbitmqctl set_permissions -p shop orders '^orders.*' '^orders$' '^(fulfilment|orders.*)$'
#                                          configure     write       read     (regular expressions on names)
rabbitmqctl list_queues -p shop name type messages messages_unacknowledged consumers
rabbitmq-diagnostics status

Les trois motifs de permission sont des expressions régulières sur les noms de ressources, pour configure (déclarer et supprimer), write (publier vers un exchange) et read (consommer une file, lier une file). L'outil en ligne de commande rabbitmqadmin encapsule l'API HTTP pour scripter les déclarations, et rabbitmq-diagnostics rend compte de la santé des nœuds, des alarmes et de l'état du cluster. La première alarme que vous rencontrerez en pratique est l'alarme de ressources mémoire ou disque : lorsqu'un nœud dépasse son seuil haut (high watermark), il bloque toutes les connexions qui publient jusqu'à ce que les consommateurs rattrapent leur retard, ce qui ressemble à un blocage pur et simple vu du producteur.

Erreurs courantes

• Publier vers un nom de file sur un exchange personnalisé. La routing key doit correspondre à un binding ; sans binding, le message est jeté et rien n'est journalisé. Utilisez mandatory avec un return listener pendant le développement.
• Une connexion par message, ou un channel partagé entre threads. La première épuise les descripteurs de fichiers et le broker ; le second corrompt le flux protocolaire. Une connexion par processus, un channel par thread.
• File durable, message transitoire. Un delivery mode 1 sur une file classique ne survit pas à un redémarrage, même si la file, elle, survit. Utilisez le mode 2, ou des quorum queues, qui ignorent cette distinction.
• Pas de prefetch. Un consommateur prend tout, les autres restent inactifs, et un plantage entraîne la relivraison de milliers de messages. Appelez toujours basicQos.
• Boucles de remise en file infinies. Un message empoisonné nacké avec requeue = true revient aussitôt. Plafonnez les relivraisons avec x-delivery-limit ou une vérification de x-death, et envoyez-le en dead letter.
• Figer des réglages changeants dans les arguments de file. Les TTL, limites et noms de DLX évoluent ; les arguments de file, non. Utilisez des policies.

Points clés à retenir

• Les producteurs envoient aux exchanges avec une routing key ; les bindings décident quelles files reçoivent une copie ; les consommateurs lisent les files. Rien d'autre dans RabbitMQ n'a de sens sans ce modèle.
• Les exchanges direct comparent les clés à l'identique, fanout copie vers tout le monde, topic applique des motifs * et #, headers compare des attributs ; un exchange topic par contexte avec une convention de routing key couvre la plupart des conceptions.
• Les quorum queues sont le choix par défaut pour le travail durable ; les files classiques servent les usages transitoires et exclusifs ; les streams ajoutent des journaux rejouables.
• La livraison at-least-once exige tout à la fois exchange durable, file durable, message persistant, publisher confirms et acquittements manuels.
• Le prefetch de basicQos garde les consommateurs équitables et rend les plantages peu coûteux ; les consumer timeouts et les delivery limits empêchent les messages bloqués ou empoisonnés de faire tomber le broker.
• Exprimez les TTL, limites de longueur et routages dead-letter sous forme de policies, pour pouvoir les modifier sans redéclarer les files.