# Dev Rabbitmq Patterns Guide > Patterns RabbitMQ — exchanges (direct, topic, fanout, headers), routing keys, dead letter, TTL, quorum queues, acks et intégration MassTransit/.NET. Spécifique à RabbitMQ. Se déclenche avec "RabbitMQ", "exchange", "routing key", "dead letter", "MassTransit", "AMQP", "quorum queue", "prefetch". Also triggers on "RabbitMQ exchange", "dead letter queue", "AMQP routing". - Skill: `khalilbenaz/dev-rabbitmq-patterns-guide` (Agent Skill) - Install (CLI): `npx skillmds@latest add khalilbenaz/dev-rabbitmq-patterns-guide` - Raw SKILL.md: https://api.skillmd.com/api/skills/khalilbenaz/dev-rabbitmq-patterns-guide/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - Author: khalilbenaz (https://skillmd.com/u/khalilbenaz) - Updated: 2026-09-21 - Page: https://skillmd.com/skills/khalilbenaz/dev-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 | Exchange | Routing | Choisir quand… | |----------|---------|----------------| | **Direct** | Routing key exacte | File de commandes ciblées, work queues | | **Fanout** | Broadcast toutes queues | Notification multi-services, cache invalidation | | **Topic** | Pattern `*.order.#` | Routing par catégorie/domaine flexible | | **Headers** | Attributs headers | Critères multiples sans modifier la routing key | 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 ``` ```bash # CLI — déclarer la queue durable rabbitmqadmin declare queue name=jobs durable=true arguments='{"x-dead-letter-exchange":"dlx.jobs"}' ``` ```csharp // 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 ```csharp builder.Services.AddMassTransit(x => { x.AddConsumer(); x.AddConsumer(); 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(context); }); cfg.ConfigureEndpoints(context); }); }); ``` ### Consumer idempotent ```csharp public class OrderCreatedConsumer : IConsumer { private readonly IOrderRepository _repo; public OrderCreatedConsumer(IOrderRepository repo) => _repo = repo; public async Task Consume(ConsumeContext 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 ```csharp // É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 ```bash # 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 | Métrique | Seuil d'alerte | Action corrective | |----------|---------------|-------------------| | Queue depth | > 10 000 | Ajouter des consumers / investiguer lenteur | | Consumer lag | > 5 min | Profiler le consumer, check DB | | DLQ count | > 0 | Analyser le message en erreur, rejouer si corrigé | | Connection drops | > 1/h | Vérifier heartbeat, pare-feu, timeouts load balancer | | Memory usage broker | > 80 % | Activer flow control, purger ou scaler | | Unacked messages | Stables > 5 min | Consumer bloqué ou crash → redémarrer | ```bash # 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 | Anti-pattern | Symptôme | Correction | |---|---|---| | ACK avant traitement | Perte de messages en cas de crash | ACK uniquement après persistance réussie | | Queue non durable | Messages perdus au redémarrage broker | `durable=true` + messages persistent | | Pas de DLX | Messages disparus silencieusement | Toujours lier une DLX à chaque queue | | Consumer non idempotent | Traitements en double après retry | Clé d'idempotence en base ou cache distribué | | Trop de prefetchCount | Consumer submergé, mémoire OOM | Commencer à 1, augmenter prudemment | | Exchange fanout pour commandes | Plusieurs services traitent la même commande | Fanout = événements uniquement ; commande → direct | | Stocker l'état dans le message | Messages géants, versioning cauchemar | Ne passer que des IDs, requêter la source | | RPC via RabbitMQ à grande échelle | Couplage temporel, timeouts en cascade | Remplacer par saga / choreography asynchrone | | Pas de heartbeat configuré | Connexions zombies non détectées | `RequestedHeartbeat = 30s` côté client | | Connexion partagée entre threads | Race conditions, channel errors | Une connexion par process, un channel par thread | --- ## Bonnes pratiques 2026 - **Quorum queues** en production : remplacent les mirrored queues (dépréciées) pour la haute disponibilité. ```bash 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. ## Communication Rules — MANDATORY - Ultra-concise. No filler, no preamble, no pleasantries. - Never say "happy to help", "sure!", "great question", "let me", or similar. - Tool first, talk second. Act before explaining. - Result first. Lead with outcome, not process. - Stop when done. No summary, no recap, no trailing commentary. - No politeness wrappers. Direct and blunt. - Minimum words. If one word works, do not use ten. - No unsolicited explanations. - No emoji unless asked.