Rabbitmq Patterns Guide

Guide des Patterns RabbitMQ

Workflow en 4 étapes

1. Identifier le pattern : point-à-point (work queue), pub/sub (fanout), routing ciblé (direct/topic), RPC ou Saga.
2. Concevoir la topologie : choisir le type d'exchange, nommer les queues et les routing keys, prévoir les DLX dès le départ.
3. Implémenter : producteur avec confirm mode, consumer avec ACK manuel et idempotence.
4. Fiabiliser : configurer retry + dead letter, activer la persistence, mettre en place le monitoring.

Critères de choix d'exchange

Règle : si tu te retrouves à créer plusieurs Direct exchanges pour le même producteur, passe en Topic.

Patterns fondamentaux

1. Work Queue (répartition de charge)

Producer → [exchange direct] → [queue:jobs] → Consumer 1
                                             → Consumer 2

# CLI — déclarer la queue durable
rabbitmqadmin declare queue name=jobs durable=true arguments='{"x-dead-letter-exchange":"dlx.jobs"}'

// Producteur : confirm mode obligatoire pour ne pas perdre de messages
var factory = new ConnectionFactory { HostName = "localhost" };
using var connection = await factory.CreateConnectionAsync();
using var channel = await connection.CreateChannelAsync();

await channel.QueueDeclareAsync("jobs", durable: true, exclusive: false, autoDelete: false);
await channel.ConfirmSelectAsync();

var body = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(payload));
var props = new BasicProperties { Persistent = true };
await channel.BasicPublishAsync("", "jobs", props, body);
await channel.WaitForConfirmsOrDieAsync(); // lève une exception si NACK

• prefetchCount = 1 : chaque consumer reçoit un message à la fois → répartition équitable.
• ACK manuel après traitement réussi ; NACK + requeue=false si l'erreur est définitive.

2. Pub/Sub (fanout, diffusion d'événements)

Producer → [exchange:events.fanout] → [queue:svc-A] → Consumer A
                                    → [queue:svc-B] → Consumer B

Chaque service crée sa propre queue et la bind à l'exchange. L'exchange copie le message vers toutes les queues liées.

3. Topic Routing

Producer → [exchange:domain.topic]
    routing key "order.created.fr"  → queue:orders-fr
    routing key "payment.failed.*"  → queue:alerts
    routing key "#.audit"           → queue:audit-all

• * = exactement un mot.
• # = zéro ou plusieurs mots.

4. RPC (requête / réponse synchrone via RabbitMQ)

À éviter si possible (couplage temporel). Préférer un pattern saga asynchrone. Si nécessaire : le producteur envoie dans une queue avec reply_to = queue de réponse temporaire et correlation_id = UUID. Le consumer publie la réponse dans reply_to.

Intégration MassTransit (.NET)

Configuration complète

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderCreatedConsumer>();
    x.AddConsumer<PaymentFailedConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost", h =>
        {
            h.Username("guest");
            h.Password("guest");
            // Heartbeat pour détecter les connexions mortes
            h.RequestedHeartbeat = TimeSpan.FromSeconds(30);
        });

        cfg.ReceiveEndpoint("order-service", e =>
        {
            e.PrefetchCount = 16;

            // Retry immédiat (erreurs réseau, DB transitoire)
            e.UseMessageRetry(r => r.Intervals(
                TimeSpan.FromSeconds(1),
                TimeSpan.FromSeconds(5),
                TimeSpan.FromSeconds(30)));

            // Relivraison différée (dépendance externe lente)
            e.UseDelayedRedelivery(r => r.Intervals(
                TimeSpan.FromMinutes(1),
                TimeSpan.FromMinutes(5),
                TimeSpan.FromMinutes(30)));

            e.ConfigureConsumer<OrderCreatedConsumer>(context);
        });

        cfg.ConfigureEndpoints(context);
    });
});

Consumer idempotent

public class OrderCreatedConsumer : IConsumer<OrderCreated>
{
    private readonly IOrderRepository _repo;

    public OrderCreatedConsumer(IOrderRepository repo) => _repo = repo;

    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        // Idempotence : court-circuit si déjà traité
        if (await _repo.ExistsAsync(context.Message.OrderId))
            return;

        await _repo.ProcessOrderAsync(context.Message);

        // Publier l'événement suivant dans la même transaction logique
        await context.Publish(new OrderProcessed
        {
            OrderId = context.Message.OrderId,
            ProcessedAt = DateTime.UtcNow
        });
    }
}

Contrats de messages

// Événement : passé accompli, immutable, versioable
public record OrderCreated
{
    public Guid OrderId { get; init; }
    public string CustomerId { get; init; } = default!;
    public decimal Amount { get; init; }
    public DateTime CreatedAt { get; init; }
    public int SchemaVersion { get; init; } = 1; // versionning explicite
}

// Commande : impératif, un seul consommateur cible
public record ProcessPayment
{
    public Guid PaymentId { get; init; }
    public Guid OrderId { get; init; }
    public decimal Amount { get; init; }
}

Dead Letter & Retry

Flux de retry recommandé

Message → Queue principale → Consumer
              ↓ NACK (erreur transitoire)
         Retry immédiat x3 (1s / 5s / 30s)
              ↓ encore en échec
         Delayed redelivery x3 (1min / 5min / 30min)
              ↓ encore en échec
         Dead Letter Queue (DLQ) → alerte Slack/PagerDuty

Déclarer la DLX manuellement

# Exchange DLX + queue DLQ
rabbitmqadmin declare exchange name=dlx.orders type=direct durable=true
rabbitmqadmin declare queue name=dlq.orders durable=true
rabbitmqadmin declare binding source=dlx.orders destination=dlq.orders routing_key=orders

# Queue principale avec DLX attaché
rabbitmqadmin declare queue name=orders durable=true \
  arguments='{"x-dead-letter-exchange":"dlx.orders","x-message-ttl":86400000}'

Monitoring & opérations

# Rejouer les messages de la DLQ vers la queue principale (shovel one-shot)
rabbitmqctl set_parameter shovel replay \
  '{"src-protocol":"amqp091","src-uri":"amqp://","src-queue":"dlq.orders",
    "dest-protocol":"amqp091","dest-uri":"amqp://","dest-queue":"orders",
    "src-delete-after":"queue-length"}'

Anti-patterns et pièges

Bonnes pratiques 2026

• Quorum queues en production : remplacent les mirrored queues (dépréciées) pour la haute disponibilité.

  rabbitmqadmin declare queue name=orders durable=true \
    arguments='{"x-queue-type":"quorum","x-dead-letter-exchange":"dlx.orders"}'
  

• Lazy queues (x-queue-mode: lazy) pour les queues volumineuses à faible débit — stocke sur disque, préserve la RAM.
• Vhost par domaine : isoler orders, payments, notifications dans des vhosts séparés.
• Connection pooling côté client : une connexion TCP, plusieurs channels (légers).
• Schema Registry ou versionning explicite des contrats (SchemaVersion) pour éviter les désynchronisations inter-services.
• Outbox pattern côté producteur : persister l'événement dans la même transaction DB, puis le publier de manière asynchrone — garantit l'exactly-once emission sans XA.
• Consumer scaling horizontal : stateless + idempotent = ajout de replicas sans interruption.