# Publish/Subscribe Pattern With RabbitMQ in .NET > The Publish/Subscribe pattern decouples message producers from consumers. Here is how to implement it with RabbitMQ and MassTransit in .NET, including retries, the outbox pattern, and idempotent consumers. Published: 2026-10-27. Author: Milan Jovanović. Canonical: https://milanjovanovic.tech/blog/pub-sub-pattern-rabbitmq-dotnet **Publish/subscribe** gives each subscriber its own copy of an event through a separate queue. In RabbitMQ, MassTransit creates message exchanges and subscriber bindings, while independent consumers handle their own backlog. Durability requires pre-existing queues, confirmed publication, and idempotent processing; an outbox coordinates events with database changes. You place an order, and three things need to happen: the shipping service creates a shipment, the notification service sends an email, and the analytics service records the sale. If the order service calls all three over HTTP, it is coupled to every one of them. Add a fourth consumer, and you change the order service again. If one call fails, the whole request fails. The **Publish/Subscribe pattern** solves this. The order service publishes one event and moves on. Any number of subscribers react to it independently. ## What Is Pub/Sub? In the Publish/Subscribe pattern, a publisher sends a message without knowing who will receive it. Subscribers independently decide which messages they're interested in. The message broker (RabbitMQ) handles routing. This is different from point-to-point messaging where a producer sends to a specific queue. With pub/sub, one message can reach multiple subscribers. I cover the other queueing patterns in [**message queue patterns in .NET**](https://milanjovanovic.tech/blog/message-queue-patterns-dotnet). ## RabbitMQ Concepts - **Exchange**: Receives messages from publishers and routes them to queues - **Queue**: Stores messages until a consumer processes them - **Binding**: Rules that connect exchanges to queues - **Consumer**: Reads and processes messages from a queue A **fanout exchange** sends every message to all bound queues (classic pub/sub). If you're brand new to the broker, start with my [**RabbitMQ beginner's guide**](https://milanjovanovic.tech/blog/rabbitmq-dotnet-beginners-guide) for the core mechanics. ![Order service publishing to a fanout exchange that copies each message into the shipping, notification, and analytics queues, each drained by its own consumer](https://milanjovanovic.tech/blogs/articles/pub-sub-pattern-rabbitmq-dotnet/fanout-exchange.png) ## Setting Up MassTransit This example runs shipping and notification consumers in one application so you can inspect both queues locally. Deploy them as separate services when they need independent ownership or scaling. It targets MassTransit 8.5.11; the [**getting-started guide**](https://milanjovanovic.tech/blog/masstransit-dotnet-getting-started) covers the v8/v9 licensing distinction. ```bash docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-management docker run -d --name pubsub-postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=shipping -p 5432:5432 postgres:17-alpine dotnet new web -n PubSubDemo --framework net10.0 cd PubSubDemo dotnet add package MassTransit.RabbitMQ --version 8.5.11 dotnet add package Npgsql.EntityFrameworkCore.PostgreSQL --version 10.0.0 ``` Use these disposable local credentials only for the demo. Create `Program.cs`; the consumer and database types below go in separate files: ```csharp using MassTransit; using Microsoft.EntityFrameworkCore; using PubSubDemo.Contracts; var builder = WebApplication.CreateBuilder(args); builder.Services.AddDbContext(o => o.UseNpgsql( "Host=localhost;Database=shipping;Username=postgres;Password=postgres")); builder.Services.AddMassTransit(x => { x.SetKebabCaseEndpointNameFormatter(); x.AddConsumer(); x.AddConsumer(); x.UsingRabbitMq((context, cfg) => { cfg.Host("localhost"); cfg.UseMessageRetry(r => r.Immediate(2)); cfg.ConfigureEndpoints(context); }); }); builder.Services.Configure(o => o.WaitUntilStarted = true); var app = builder.Build(); await using (var scope = app.Services.CreateAsyncScope()) { var db = scope.ServiceProvider.GetRequiredService(); await db.Database.EnsureCreatedAsync(); } app.MapPost("/events/orders", async ( OrderPlacedEvent message, IPublishEndpoint publisher, CancellationToken ct) => { await publisher.Publish(message, ct); return Results.Accepted(); }); app.Run(); ``` `EnsureCreatedAsync` is only for this disposable database. Use reviewed migrations in a persistent environment, and do not mix `EnsureCreated` with migrations for the same database. ## Define the Message Contract ```csharp // Contracts.cs, referenced by publisher and subscribers namespace PubSubDemo.Contracts; public sealed record OrderPlacedEvent { public Guid OrderId { get; init; } public Guid CustomerId { get; init; } public decimal TotalAmount { get; init; } public List Items { get; init; } = []; public DateTime OccurredAt { get; init; } } public sealed record OrderItemDto { public Guid ProductId { get; init; } public int Quantity { get; init; } public decimal UnitPrice { get; init; } } ``` Message contracts are plain classes or records. Put them in a shared NuGet package or contracts assembly. Treat contracts as immutable once published. Renaming a property is a breaking change for every subscriber, and they don't all deploy at the same time. Only make additive changes (new optional properties), or introduce a new event version. ## Publishing Messages The `/events/orders` endpoint publishes an existing `OrderPlacedEvent`; it does not save an order. Supply a stable `OrderId`, `CustomerId`, and `TotalAmount` in the JSON body and post it to the URL printed by `dotnet run`. `IPublishEndpoint.Publish` routes to the message type's exchange, then to bound subscriber queues. MassTransit creates those bindings when the receive endpoints start. Publishing before a subscriber has ever started can lose that subscriber's copy, even though the broker accepted the event. Provision topology before accepting production traffic. ## Subscribing to Messages Each subscriber implements `IConsumer` for the events it cares about. ### Subscriber 1: Shipping Service Create `ShippingDbContext.cs` with `OrderId` as the shipment's unique business key: ```csharp using Microsoft.EntityFrameworkCore; public sealed class ShippingDbContext(DbContextOptions options) : DbContext(options) { public DbSet Shipments => Set(); protected override void OnModelCreating(ModelBuilder modelBuilder) { modelBuilder.Entity().HasKey(s => s.OrderId); } } public sealed class Shipment { public Guid OrderId { get; set; } public Guid CustomerId { get; set; } } ``` Create `OrderPlacedShippingConsumer.cs`: ```csharp using MassTransit; using Microsoft.EntityFrameworkCore; using PubSubDemo.Contracts; public sealed class OrderPlacedShippingConsumer(ShippingDbContext db) : IConsumer { public async Task Consume(ConsumeContext context) { var message = context.Message; if (message.OrderId == Guid.Empty || message.CustomerId == Guid.Empty) throw new ArgumentException("Order and customer IDs are required."); await db.Database.ExecuteSqlInterpolatedAsync($""" INSERT INTO "Shipments" ("OrderId", "CustomerId") VALUES ({message.OrderId}, {message.CustomerId}) ON CONFLICT ("OrderId") DO NOTHING """, context.CancellationToken); } } ``` The PostgreSQL primary key and `ON CONFLICT` make concurrent duplicate deliveries harmless for this one-row effect. The interpolated EF Core API parameterizes the values. This assumes one immutable shipment request per order; more complex workflows need an inbox transaction and explicit handling of conflicting payloads. ### Subscriber 2: Notification Service Create `OrderPlacedNotificationConsumer.cs`: ```csharp using MassTransit; using Microsoft.Extensions.Logging; using PubSubDemo.Contracts; public sealed class OrderPlacedNotificationConsumer( ILogger logger) : IConsumer { public Task Consume(ConsumeContext context) { logger.LogInformation("Confirmation requested for {OrderId}", context.Message.OrderId); return Task.CompletedTask; } } ``` This consumer logs the notification request; it does not send an email. When connecting an email provider, use its idempotency mechanism or persist a separate delivery job, because a database transaction cannot atomically commit an external email. `ConfigureEndpoints` creates separate queues for these two consumers. An existing durable notification queue retains its backlog during an outage, subject to configured retention and capacity limits, while shipping can continue. ## Error Handling and Retries The initial registration enables two immediate retries. To use a longer policy, replace that call with the following configuration: ```csharp x.UsingRabbitMq((context, cfg) => { cfg.Host("localhost"); cfg.UseMessageRetry(r => { r.Exponential( retryLimit: 3, minInterval: TimeSpan.FromSeconds(1), maxInterval: TimeSpan.FromSeconds(30), intervalDelta: TimeSpan.FromSeconds(5)); }); cfg.ConfigureEndpoints(context); }); ``` After exhausting retries, messages go to an error queue (`_error` suffix) for investigation. That error queue is effectively a [**dead letter queue**](https://milanjovanovic.tech/blog/dead-letter-queues-failed-messages), and you should monitor it. A growing error queue means a consumer is broken. ## Duplicate Messages and Idempotency A consumer can commit its work and crash before acknowledging the message. Another instance then receives the same event. Checking `AnyAsync` before inserting does not prevent two concurrent instances from both passing the check. The shipment example uses a database-enforced primary key and an atomic insert. When a consumer updates several tables, write an [**inbox record**](https://milanjovanovic.tech/blog/implementing-the-inbox-pattern-for-reliable-message-consumption) in the same database transaction as those changes, with a unique constraint on the message ID. Do not commit the inbox record before the business work succeeds. ## Publish Database Changes Through an Outbox The endpoint above intentionally accepts an already-created event to demonstrate routing. In an order-creation endpoint, saving an order and then calling the broker creates a failure gap: a crash can leave the order committed without its event. The [**MassTransit bus outbox**](https://masstransit.massient.com/concepts/outbox) captures messages through a scoped publisher and persists them with the application's database changes. Register the EF outbox entities, configure `UsePostgres` and `UseBusOutbox`, publish the complete contract, then call `SaveChangesAsync` once in that same scope. Omitting `CustomerId` or line items from the outbox event changes what subscribers receive even if transport delivery is reliable. The [**MassTransit getting-started guide**](https://milanjovanovic.tech/blog/masstransit-dotnet-getting-started) contains the context mapping and registration together. For the underlying transaction and delivery loop, see [**transactional outbox with EF Core**](https://milanjovanovic.tech/blog/transactional-outbox-ef-core). ## When NOT to Use Pub/Sub Pub/sub is not a default for all communication: - **The caller needs a response.** Pub/sub is fire-and-forget. If placing an order must return a payment result, that's a request/response interaction, not an event. - **The steps form a workflow with compensation.** If a downstream failure must undo upstream work, use a [**saga**](https://milanjovanovic.tech/blog/saga-pattern-dotnet) instead of a chain of loosely observed events. - **There's exactly one consumer and there always will be.** A plain queue is simpler to reason about. - **You need strict ordering across subscribers.** Fanout gives each queue its own copy, and consumers process at their own pace. Events also make the workflow harder to follow. When the "what happens after an order" logic is spread across five subscribers, no single place in the code shows the full picture. Pay that cost only when the decoupling buys you something. ## Testing Subscribers Independently Test shipping against PostgreSQL: publish the same order twice, including concurrent deliveries, and assert that only one shipment exists. EF Core's InMemory provider cannot verify PostgreSQL `ON CONFLICT` or unique constraints. Use the [**MassTransit test harness**](https://masstransit.massient.com/guides/unit-testing/consumer) for consumer dispatch, then assert the committed database result and inspect consumer exceptions. Test RabbitMQ routing separately by running the two subscribers, stopping notifications, publishing an event, and checking that shipping completes while the notification queue retains its copy. A harness with an in-memory transport does not validate RabbitMQ topology or durability. ## Summary Subscriber isolation is what makes pub/sub useful for [**event-driven architectures**](https://milanjovanovic.tech/blog/event-driven-architecture-in-dotnet-with-rabbitmq). Give every subscriber its own durable queue so one consumer's outage or retry policy cannot block another. Publish through an outbox, make each consumer idempotent, and monitor error queues as part of the service that owns the subscription. ## Frequently asked questions ### What is the difference between pub/sub and a message queue? A queue distributes delivery attempts among its consumers (point-to-point); failures can cause redelivery. Pub/sub delivers a copy of each message to every subscriber. In RabbitMQ, pub/sub is implemented with a fanout or topic exchange bound to multiple queues, one per subscriber. ### Do I need MassTransit to do pub/sub with RabbitMQ in .NET? No, you can use the raw RabbitMQ.Client library and declare exchanges and bindings yourself. MassTransit saves you from writing that plumbing: it creates exchanges and queues by convention, handles serialization, retries, and error queues. ### What happens if a subscriber is down when a message is published? Each subscriber has its own durable queue, so messages wait in the queue until the subscriber comes back online. The queue and binding must already exist, publication must be confirmed, and retention or queue limits can still remove messages. ### How do I avoid processing the same message twice? With reliable publishing and manual acknowledgments, deliveries can repeat after failure. Make consumers idempotent: enforce a unique business key or persist an inbox record atomically with the business change. ### Why do I need the outbox pattern with pub/sub? Publishing directly after a database save is two separate operations, and either can fail independently. The outbox pattern stores the message in the same database transaction as your data, and a background process publishes it afterwards.