RabbitMQ Study Portal

How it works

MassTransit setup

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderPlacedConsumer>();
    x.AddConsumer<TenantCreatedConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration["RabbitMq:Host"], "/", h =>
        {
            h.Username(builder.Configuration["RabbitMq:Username"]!);
            h.Password(builder.Configuration["RabbitMq:Password"]!);
        });

        cfg.ConfigureEndpoints(context);   // auto-declares exchanges, queues, bindings
    });
});

ConfigureEndpoints reads every registered consumer, derives a queue per consumer, declares the exchanges and bindings for every message type the consumer accepts. This is the messaging-equivalent of MigrateAsync — topology is applied on every startup, idempotently.

Publishing (type-safe contracts)

public record OrderPlaced(Guid OrderId, decimal Amount, DateTime PlacedAt);

public class OrderService
{
    private readonly IPublishEndpoint _bus;
    public OrderService(IPublishEndpoint bus) => _bus = bus;

    public async Task PlaceOrder(Order order, CancellationToken ct)
    {
        await _orderRepo.SaveAsync(order, ct);
        await _bus.Publish(new OrderPlaced(order.Id, order.Amount, DateTime.UtcNow), ct);
    }
}

Publish<T> routes to an exchange named after the type's full name. Every consumer of that type gets a copy (fan-out semantics).

For point-to-point (one consumer), use Send to a specific endpoint:

var endpoint = await _bus.GetSendEndpoint(new Uri("queue:order-processor"));
await endpoint.Send(new ProcessOrder(orderId), ct);

Consuming

public class OrderPlacedConsumer : IConsumer<OrderPlaced>
{
    private readonly INotificationService _notifications;

    public OrderPlacedConsumer(INotificationService notifications) =>
        _notifications = notifications;

    public async Task Consume(ConsumeContext<OrderPlaced> ctx)
    {
        await _notifications.SendOrderConfirmation(ctx.Message.OrderId, ctx.CancellationToken);
    }
}

Consumers are resolved per message via DI. Exceptions thrown from Consume trigger retry/redelivery (see below). Successful return acks the message.

Retry and redelivery

cfg.ReceiveEndpoint("order-placed-queue", e =>
{
    e.UseMessageRetry(r => r.Intervals(
        TimeSpan.FromSeconds(1),
        TimeSpan.FromSeconds(5),
        TimeSpan.FromSeconds(30)
    ));

    e.UseScheduledRedelivery(r => r.Intervals(
        TimeSpan.FromMinutes(5),
        TimeSpan.FromMinutes(30),
        TimeSpan.FromHours(1)
    ));

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

• Retry = immediate in-memory retries within the consumer instance
• ScheduledRedelivery = puts the message back on the queue with a delay (uses RabbitMQ delayed-message exchange plugin)
• After all retries exhaust → message moves to <queue>_error

Sagas (long-running workflows)

public class OrderState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; } = null!;
    public decimal Amount { get; set; }
    public DateTime PlacedAt { get; set; }
}

public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public State AwaitingPayment { get; private set; } = null!;
    public State Paid { get; private set; } = null!;

    public Event<OrderPlaced> OrderPlaced { get; private set; } = null!;
    public Event<PaymentReceived> PaymentReceived { get; private set; } = null!;

    public OrderStateMachine()
    {
        InstanceState(x => x.CurrentState);
        Event(() => OrderPlaced, x => x.CorrelateById(c => c.Message.OrderId));
        Event(() => PaymentReceived, x => x.CorrelateById(c => c.Message.OrderId));

        Initially(
            When(OrderPlaced)
                .Then(ctx => { ctx.Saga.Amount = ctx.Message.Amount; })
                .TransitionTo(AwaitingPayment));

        During(AwaitingPayment,
            When(PaymentReceived).TransitionTo(Paid).Finalize());
    }
}

State is persisted (EF Core / Marten / Redis). Saga survives restarts. Use for any workflow where you need to "remember where we are."

Outbox pattern (transactional consistency with EF Core)

builder.Services.AddMassTransit(x =>
{
    x.AddEntityFrameworkOutbox<MyDbContext>(o =>
    {
        o.QueryDelay = TimeSpan.FromSeconds(1);
        o.UsePostgres();
        o.UseBusOutbox();
    });

    x.AddConsumer<OrderPlacedConsumer>();
    x.UsingRabbitMq((ctx, cfg) => cfg.ConfigureEndpoints(ctx));
});

Now Publish inside the same DbContext transaction is captured into an OutboxMessage table — published to RabbitMQ only after SaveChanges commits. Solves dual-write inconsistency (DB update succeeds, broker publish fails, or vice versa).

Scheduled messages

await _bus.SchedulePublish(
    DateTime.UtcNow.AddMinutes(15),
    new ReminderTrigger(userId),
    context.CancellationToken);

Uses Quartz.NET in-process or the RabbitMQ delayed-message-exchange plugin (preferred).

Health checks

builder.Services.AddHealthChecks()
    .AddRabbitMQ(rabbitConnectionString, tags: new[] { "ready" });

/health/ready returns 503 if RabbitMQ is unreachable. Wire into Kubernetes readiness probes.