MassTransit is an abstraction over message brokers — RabbitMQ, Azure Service Bus, Amazon SQS, and others. It provides publish/subscribe, request/response, and saga-based workflows without tying your code to a specific transport. You swap brokers by changing configuration, not rewriting consumers.

Setup with RabbitMQ

dotnet add package MassTransit.RabbitMQ

Configure in Program.cs:

Program.cs 
builder.Services.AddMassTransit(x =>
{
    x.AddConsumersFromNamespaceContaining<OrderPlacedConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

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

ConfigureEndpoints automatically creates queues and bindings based on your consumers' names.

Defining Messages

Messages are plain C# types — records work well:

Example.cs 
public record OrderPlaced(Guid OrderId, string CustomerEmail, decimal Total);

public record OrderShipped(Guid OrderId, string TrackingNumber);

public record OrderCancelled(Guid OrderId, string Reason);

MassTransit uses the type name for routing. Keep messages in a shared contracts assembly that both publisher and consumer reference.

Publishing Events

Inject IPublishEndpoint or IBus to publish messages:

Example.cs 
public class OrderService
{
    private readonly IPublishEndpoint _publishEndpoint;

    public OrderService(IPublishEndpoint publishEndpoint)
    {
        _publishEndpoint = publishEndpoint;
    }

    public async Task PlaceOrderAsync(Order order, CancellationToken ct)
    {
        await _repository.SaveAsync(order, ct);

        await _publishEndpoint.Publish(new OrderPlaced(
            order.Id,
            order.CustomerEmail,
            order.Total), ct);
    }
}

Publishing delivers the message to all consumers subscribed to that message type (fan-out).

Consuming Messages

A consumer handles a specific message type:

Example.cs 
public class OrderPlacedConsumer : IConsumer<OrderPlaced>
{
    private readonly IEmailService _emailService;
    private readonly ILogger<OrderPlacedConsumer> _logger;

    public OrderPlacedConsumer(
        IEmailService emailService,
        ILogger<OrderPlacedConsumer> logger)
    {
        _emailService = emailService;
        _logger = logger;
    }

    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        _logger.LogInformation("Order {OrderId} placed, sending confirmation",
            context.Message.OrderId);

        await _emailService.SendOrderConfirmationAsync(
            context.Message.CustomerEmail,
            context.Message.OrderId);
    }
}

Multiple consumers can handle the same message type. Each gets its own queue, so every consumer receives a copy.

Sending Commands

Commands go to a specific endpoint (point-to-point), unlike events which fan out:

Example.cs 
public record ProcessPayment(Guid OrderId, decimal Amount, string Currency);

// Sending a command
var endpoint = await _bus.GetSendEndpoint(
    new Uri("queue:process-payment"));

await endpoint.Send(new ProcessPayment(orderId, 49.99m, "GBP"));

Consumer Configuration

Configure retries, concurrency, and error handling per consumer:

Example.cs 
x.UsingRabbitMq((context, cfg) =>
{
    cfg.Host("localhost");

    cfg.ReceiveEndpoint("order-placed", e =>
    {
        e.ConfigureConsumer<OrderPlacedConsumer>(context);

        e.UseMessageRetry(r => r
            .Incremental(3, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2)));

        e.PrefetchCount = 16;
        e.ConcurrentMessageLimit = 8;
    });
});

Request/Response

MassTransit supports request/response for synchronous-style communication over async messaging:

Example.cs 
public record GetOrderStatus(Guid OrderId);
public record OrderStatusResult(Guid OrderId, string Status);

public class GetOrderStatusConsumer : IConsumer<GetOrderStatus>
{
    public async Task Consume(ConsumeContext<GetOrderStatus> context)
    {
        var status = await _repository.GetStatusAsync(context.Message.OrderId);

        await context.RespondAsync(new OrderStatusResult(
            context.Message.OrderId, status));
    }
}

// Client side
public class OrderClient
{
    private readonly IRequestClient<GetOrderStatus> _client;

    public OrderClient(IRequestClient<GetOrderStatus> client)
    {
        _client = client;
    }

    public async Task<string> GetStatusAsync(Guid orderId, CancellationToken ct)
    {
        var response = await _client.GetResponse<OrderStatusResult>(
            new GetOrderStatus(orderId), ct);

        return response.Message.Status;
    }
}

Error Handling

When a consumer throws, MassTransit retries based on your configuration. After exhausting retries, the message goes to an error queue (_error suffix by default). You can also configure a dead-letter queue or a fault consumer:

Example.cs 
public class OrderPlacedFaultConsumer : IConsumer<Fault<OrderPlaced>>
{
    public async Task Consume(ConsumeContext<Fault<OrderPlaced>> context)
    {
        _logger.LogError("Failed to process OrderPlaced for {OrderId}: {Errors}",
            context.Message.Message.OrderId,
            string.Join(", ", context.Message.Exceptions.Select(e => e.Message)));
    }
}

Testing

MassTransit includes an in-memory test harness:

Example.cs 
[Fact]
public async Task Should_consume_order_placed()
{
    await using var provider = new ServiceCollection()
        .AddMassTransitTestHarness(x =>
        {
            x.AddConsumer<OrderPlacedConsumer>();
        })
        .BuildServiceProvider(true);

    var harness = provider.GetRequiredService<ITestHarness>();
    await harness.Start();

    await harness.Bus.Publish(new OrderPlaced(Guid.NewGuid(), "[email protected]", 50m));

    (await harness.Consumed.Any<OrderPlaced>()).Should().BeTrue();
}

MassTransit handles the plumbing of message-based systems — serialisation, routing, retries, error queues, and consumer lifecycle. You focus on message contracts and business logic.