MassTransit для C# (.NET): полное руководство с примерами

1. Что это и зачем нужна

MassTransit — это мощная, открытая и широко используемая библиотека для создания распределённых приложений на платформе .NET. Она предоставляет высокоуровневую абстракцию над различными системами обмена сообщениями (message brokers), такими как RabbitMQ, Azure Service Bus, Amazon SQS/SNS, ActiveMQ и Kafka (через коннекторы). Основная цель MassTransit — упростить разработку сервисов, взаимодействующих через асинхронные сообщения, избавив разработчика от необходимости вникать в низкоуровневые детали конкретного брокера.

Зачем нужна MassTransit? В современной микросервисной архитектуре сервисы должны общаться друг с другом надёжно и асинхронно. Вместо прямых HTTP-вызовов (которые могут привести к каскадным сбоям и жёсткой связности) используется обмен сообщениями. MassTransit берёт на себя такие сложные задачи, как: управление подключениями и каналами, повторные попытки отправки при сбоях, обработка ошибок, распределённые транзакции (через саги), управление очередями и топиками, сериализация/десериализация сообщений, а также предоставляет удобный API для отправки, публикации и потребления сообщений.

Библиотека реализует несколько ключевых паттернов: Command-Query Separation (через отправку команд), Publish-Subscribe (через публикацию событий), Request-Response (для синхронного обмена через асинхронные сообщения) и Saga Pattern (для управления долгоживущими бизнес-процессами). Благодаря этому MassTransit является де-факто стандартом для построения надёжных, масштабируемых и тестируемых распределённых систем на .NET.

2. Установка

MassTransit устанавливается через NuGet. В зависимости от используемого брокера сообщений, вам потребуется установить основной пакет и пакет для конкретного транспорта.

Базовая установка (без транспорта):

dotnet add package MassTransit

Для RabbitMQ:

dotnet add package MassTransit.RabbitMQ

Для Azure Service Bus:

dotnet add package MassTransit.Azure.ServiceBus.Core

Для Amazon SQS:

dotnet add package MassTransit.AmazonSQS

Для Kafka:

dotnet add package MassTransit.Kafka

3. Быстрый старт — минимальный рабочий пример

Ниже приведён простейший пример отправки и получения сообщения с использованием RabbitMQ. Убедитесь, что у вас запущен RabbitMQ (например, через Docker: docker run -d --hostname my-rabbit --name some-rabbit -p 5672:5672 -p 15672:15672 rabbitmq:3-management).

using System;
using System.Threading.Tasks;
using MassTransit;

// 1. Определяем контракт сообщения
public record OrderSubmitted
{
    public Guid OrderId { get; init; }
    public string CustomerName { get; init; }
}

// 2. Определяем потребителя (Consumer)
public class OrderSubmittedConsumer : IConsumer<OrderSubmitted>
{
    public async Task Consume(ConsumeContext<OrderSubmitted> context)
    {
        var message = context.Message;
        Console.WriteLine($"[Consumer] Получен заказ: {message.OrderId} от {message.CustomerName}");
        await Task.CompletedTask;
    }
}

// 3. Настраиваем и запускаем шину
public class Program
{
    public static async Task Main()
    {
        var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
        {
            cfg.Host("localhost", "/", h =>
            {
                h.Username("guest");
                h.Password("guest");
            });

            cfg.ReceiveEndpoint("order-queue", e =>
            {
                e.Consumer<OrderSubmittedConsumer>();
            });
        });

        await busControl.StartAsync();
        Console.WriteLine("[Bus] Шина запущена. Нажмите любую клавишу для отправки сообщения...");
        Console.ReadKey();

        // 4. Отправляем сообщение
        var endpoint = await busControl.GetSendEndpoint(new Uri("queue:order-queue"));
        await endpoint.Send(new OrderSubmitted
        {
            OrderId = Guid.NewGuid(),
            CustomerName = "Иван Иванов"
        });

        Console.WriteLine("[Sender] Сообщение отправлено. Нажмите любую клавишу для выхода...");
        Console.ReadKey();

        await busControl.StopAsync();
    }
}

4. Основные методы/функции/классы

4.1. Bus.Factory.CreateUsingRabbitMq

Сигнатура: public static IBusControl CreateUsingRabbitMq(Action<IRabbitMqBusFactoryConfigurator> configure)

Что делает: Создаёт и настраивает экземпляр шины (bus), используя RabbitMQ в качестве транспорта. Это основной метод для инициализации MassTransit с RabbitMQ.

var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    cfg.Host("localhost", "/", h =>
    {
        h.Username("guest");
        h.Password("guest");
    });
    cfg.ReceiveEndpoint("my-queue", e =>
    {
        e.Consumer<MyConsumer>();
    });
});

4.2. IBusControl.StartAsync

Сигнатура: Task StartAsync(CancellationToken cancellationToken = default)

Что делает: Запускает шину: устанавливает соединение с брокером, создаёт очереди и обменники, начинает прослушивание очередей. Должен быть вызван перед отправкой/приёмом сообщений.

await bus.StartAsync();
Console.WriteLine("Шина запущена!");

4.3. IBusControl.StopAsync

Сигнатура: Task StopAsync(CancellationToken cancellationToken = default)

Что делает: Корректно останавливает шину: завершает обработку текущих сообщений, закрывает соединения с брокером. Должен быть вызван при завершении приложения.

await bus.StopAsync();
Console.WriteLine("Шина остановлена.");

4.4. ISendEndpoint.Send

Сигнатура: Task Send<T>(T message, CancellationToken cancellationToken = default) where T : class

Что делает: Отправляет сообщение в конкретную очередь. Сообщение будет обработано одним из потребителей этой очереди (конкурирующие потребители).

var endpoint = await bus.GetSendEndpoint(new Uri("queue:order-queue"));
await endpoint.Send(new OrderSubmitted { OrderId = Guid.NewGuid(), CustomerName = "Alice" });

4.5. IPublishEndpoint.Publish

Сигнатура: Task Publish<T>(T message, CancellationToken cancellationToken = default) where T : class

Что делает: Публикует событие в шину. Все потребители, подписанные на этот тип сообщения (через свои очереди), получат его. Реализует паттерн Publish-Subscribe.

await bus.Publish(new OrderSubmitted { OrderId = Guid.NewGuid(), CustomerName = "Bob" });

4.6. IConsumer<TMessage>

Сигнатура: public interface IConsumer<in TMessage> where TMessage : class { Task Consume(ConsumeContext<TMessage> context); }

Что делает: Базовый интерфейс для всех потребителей сообщений. Ваш класс должен реализовать метод Consume, который будет вызван при получении сообщения типа TMessage.

public class MyConsumer : IConsumer<MyMessage>
{
    public async Task Consume(ConsumeContext<MyMessage> context)
    {
        var msg = context.Message;
        Console.WriteLine($"Получено: {msg.Text}");
    }
}

4.7. ConsumeContext

Сигнатура: (не создаётся напрямую, предоставляется MassTransit)

Что делает: Контекст потребления сообщения. Предоставляет доступ к самому сообщению (context.Message), а также к методам для отправки ответов (context.RespondAsync), публикации новых событий, доступа к заголовкам, повторной обработки и т.д.

public async Task Consume(ConsumeContext<OrderRequest> context)
{
    var request = context.Message;
    // Отправляем ответ обратно отправителю
    await context.RespondAsync(new OrderResponse { OrderId = request.OrderId, Status = "Processed" });
}

4.8. IRequestClient<TRequest, TResponse>

Сигнатура: public interface IRequestClient<TRequest, TResponse> where TRequest : class where TResponse : class

Что делает: Позволяет реализовать паттерн Request-Response. Клиент отправляет запрос и ожидает ответа от потребителя. Это синхронный паттерн поверх асинхронных сообщений.

// Настройка клиента
var client = bus.CreateRequestClient<OrderRequest, OrderResponse>(new Uri("queue:order-request-queue"));

// Отправка запроса и ожидание ответа
var response = await client.GetResponse<OrderResponse>(new OrderRequest { OrderId = Guid.NewGuid() });
Console.WriteLine($"Статус заказа: {response.Message.Status}");

4.9. ReceiveEndpoint

Сигнатура: (метод конфигурации: cfg.ReceiveEndpoint(string queueName, Action<IReceiveEndpointConfigurator> configure))

Что делает: Определяет точку входа для получения сообщений. Настраивает очередь, связывает её с потребителями, настраивает политики повторных попыток, количество потоков и т.д.

cfg.ReceiveEndpoint("email-queue", e =>
{
    e.Consumer<EmailConsumer>();
    e.UseMessageRetry(r => r.Interval(3, TimeSpan.FromSeconds(5))); // 3 попытки с интервалом 5 сек
    e.PrefetchCount = 10; // Количество сообщений, получаемых за раз
});

4.10. UseMessageRetry

Сигнатура: (метод расширения на IReceiveEndpointConfigurator) void UseMessageRetry(Action<IRetryConfigurator> configure)

Что делает: Настраивает политику повторных попыток для потребителя. Если потребитель выбросил исключение, MassTransit автоматически попытается обработать сообщение снова в соответствии с заданной политикой.

e.UseMessageRetry(r =>
{
    r.Interval(5, TimeSpan.FromSeconds(1)); // 5 попыток с интервалом 1 секунда
    r.Exponential(3, TimeSpan.FromMilliseconds(200), TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(5)); // Экспоненциальная задержка
    r.Incremental(3, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2)); // Увеличивающаяся задержка
});

5. Реальный практический пример (100-200 строк кода)

Рассмотрим пример системы обработки заказов в интернет-магазине. У нас есть три микросервиса: OrderService (принимает заказы), PaymentService (обрабатывает платежи) и NotificationService (отправляет уведомления). Они общаются через RabbitMQ с помощью MassTransit.

using System;
using System.Threading.Tasks;
using MassTransit;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

// ========== Контракты сообщений ==========
public record OrderCreated
{
    public Guid OrderId { get; init; }
    public decimal Amount { get; init; }
    public string CustomerEmail { get; init; }
}

public record PaymentProcessed
{
    public Guid OrderId { get; init; }
    public bool Success { get; init; }
}

public record SendNotification
{
    public string Email { get; init; }
    public string Message { get; init; }
}

// ========== OrderService (Публикует событие) ==========
public class OrderService
{
    private readonly IPublishEndpoint _publishEndpoint;

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

    public async Task CreateOrder(decimal amount, string email)
    {
        var orderId = Guid.NewGuid();
        Console.WriteLine($"[OrderService] Создан заказ {orderId} на сумму {amount}");

        // Публикуем событие о создании заказа
        await _publishEndpoint.Publish(new OrderCreated
        {
            OrderId = orderId,
            Amount = amount,
            CustomerEmail = email
        });
    }
}

// ========== PaymentService (Потребитель OrderCreated, публикует PaymentProcessed) ==========
public class PaymentConsumer : IConsumer<OrderCreated>
{
    private readonly IPublishEndpoint _publishEndpoint;

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

    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        var order = context.Message;
        Console.WriteLine($"[PaymentService] Обработка платежа для заказа {order.OrderId}");

        // Симуляция обработки платежа
        bool paymentSuccess = order.Amount < 10000; // Условно: платежи до 10000 успешны
        await Task.Delay(500); // Имитация работы

        // Публикуем результат обработки платежа
        await _publishEndpoint.Publish(new PaymentProcessed
        {
            OrderId = order.OrderId,
            Success = paymentSuccess