Wolverine для C# (.NET): полное руководство по Message Bus и CQRS

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

Wolverine — это высокопроизводительная библиотека для .NET, которая реализует паттерны Message Bus и CQRS (Command Query Responsibility Segregation). Она предоставляет разработчикам мощный и гибкий инструмент для построения распределенных, масштабируемых и отказоустойчивых приложений. В отличие от многих аналогов, Wolverine не просто абстракция над очередями сообщений, а полноценный фреймворк, который берет на себя маршрутизацию, обработку ошибок, саги, долгоживущие процессы и интеграцию с популярными транспортными протоколами (RabbitMQ, Azure Service Bus, Amazon SQS, Kafka и др.).

Зачем нужна Wolverine? В современной микросервисной архитектуре и event-driven системах часто возникает необходимость надежной передачи сообщений между компонентами. Wolverine решает эту задачу, предлагая:

Message Bus: Обеспечивает асинхронную, надежную и масштабируемую передачу сообщений между сервисами.
CQRS: Разделяет операции чтения и записи данных, что позволяет оптимизировать производительность и масштабировать каждый аспект системы независимо.
Сага (Saga): Управляет долгоживущими бизнес-процессами, которые могут состоять из нескольких шагов и требовать компенсации при сбоях.
Интеграция с ASP.NET Core: Легко встраивается в существующие проекты, используя знакомые механизмы DI и middleware.

Wolverine — это выбор для тех, кто хочет получить максимальную производительность и гибкость, не жертвуя удобством разработки. Она активно развивается сообществом и используется в production-средах.

2. Установка

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

Основная команда установки:

dotnet add package WolverineFx

Для RabbitMQ:

dotnet add package WolverineFx.RabbitMQ

Для Azure Service Bus:

dotnet add package WolverineFx.AzureServiceBus

Для Amazon SQS:

dotnet add package WolverineFx.AmazonSQS

Для Kafka:

dotnet add package WolverineFx.Kafka

После установки пакетов, настройте Wolverine в вашем Program.cs.

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

Создайте консольное приложение .NET и добавьте пакет WolverineFx. Затем замените содержимое Program.cs на следующий код:

using Wolverine;
using Microsoft.Extensions.Hosting;

// 1. Создаем хост с Wolverine
using var host = await Host.CreateDefaultBuilder(args)
    .UseWolverine(opts =>
    {
        // Настройка Wolverine (пока без транспорта, используем встроенный)
        opts.PublishAllMessages().Locally();
    })
    .StartAsync();

// 2. Получаем экземпляр IMessageBus
var bus = host.Services.GetRequiredService();

// 3. Отправляем сообщение
await bus.SendAsync(new HelloMessage("World"));

// 4. Определяем обработчик (можно в любом классе)
public class HelloMessageHandler
{
    public void Handle(HelloMessage message)
    {
        Console.WriteLine($"Hello, {message.Name}!");
    }
}

// 5. Определяем само сообщение
public record HelloMessage(string Name);

Этот минимальный пример демонстрирует отправку и обработку сообщения. Wolverine автоматически находит обработчик по типу сообщения.

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

4.1. `IMessageBus.SendAsync<T>(T message)`

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

Что делает: Отправляет сообщение асинхронно. Сообщение будет обработано одним из зарегистрированных обработчиков.

Пример:

await bus.SendAsync(new OrderCreated(123, "Laptop"));

4.2. `IMessageBus.InvokeAsync<T>(T message)`

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

Что делает: Вызывает обработчик синхронно (в том же потоке), ожидая его завершения. Полезно для тестирования или когда нужен гарантированный порядок выполнения.

Пример:

await bus.InvokeAsync(new SendEmail("user@example.com", "Welcome!"));

4.3. `IMessageBus.RequestAsync<TResponse, TRequest>(TRequest request)`

Сигнатура: Task<TResponse> RequestAsync<TResponse, TRequest>(TRequest request, CancellationToken cancellationToken = default, TimeSpan? timeout = null)

Что делает: Отправляет запрос и ожидает ответа от обработчика. Реализует паттерн Request/Response.

Пример:

var result = await bus.RequestAsync(new GetUserRequest(42));
Console.WriteLine($"User name: {result.Name}");

4.4. `IMessageBus.PublishAsync<T>(T message)`

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

Что делает: Публикует сообщение всем подписчикам. Используется для событий, когда несколько обработчиков должны реагировать на одно сообщение.

Пример:

await bus.PublishAsync(new UserRegistered("alice", "alice@example.com"));

4.5. `IMessageBus.ScheduleAsync<T>(T message, DateTimeOffset time)`

Сигнатура: Task ScheduleAsync<T>(T message, DateTimeOffset time, CancellationToken cancellationToken = default)

Что делает: Планирует отправку сообщения на определенное время в будущем.

Пример:

await bus.ScheduleAsync(new SendReminder("user@example.com"), DateTimeOffset.Now.AddHours(24));

4.6. `IMessageBus.EnqueueAsync<T>(T message)`

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

Что делает: Отправляет сообщение в локальную очередь (in-process). Полезно для асинхронной обработки в рамках одного процесса.

Пример:

await bus.EnqueueAsync(new ProcessImage("image.jpg"));

4.7. `WolverineOptions.PublishAllMessages().Locally()`

Сигнатура: PublishingExpression PublishAllMessages()

Что делает: Настраивает Wolverine на публикацию всех сообщений локально (без внешнего транспорта). Используется для тестирования или простых сценариев.

Пример:

opts.PublishAllMessages().Locally();

4.8. `WolverineOptions.Policies.AutoApplyTransactions()`

Сигнатура: PoliciesExpression AutoApplyTransactions()

Что делает: Включает автоматическое применение транзакций к обработчикам сообщений. Если обработчик использует Entity Framework, Wolverine автоматически создаст и закоммитит транзакцию.

Пример:

opts.Policies.AutoApplyTransactions();

4.9. `WolverineOptions.Publish(x => x.Message<T>()).To("rabbitmq://queue/name")`

Сигнатура: PublishingExpression Publish(Action<IPublishToExpression> configure)

Что делает: Настраивает маршрутизацию сообщений определенного типа на внешнюю очередь (например, RabbitMQ).

Пример:

opts.Publish(x => x.Message<OrderPlaced>()).To("rabbitmq://queue/orders");

4.10. `WolverineOptions.ListenForMessagesFrom("rabbitmq://queue/orders")`

Сигнатура: ListenerConfiguration ListenForMessagesFrom(Uri uri)

Что делает: Настраивает Wolverine на прослушивание сообщений из внешней очереди.

Пример:

opts.ListenForMessagesFrom("rabbitmq://queue/orders");

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

Рассмотрим пример интернет-магазина, где нужно обрабатывать заказы, отправлять уведомления и обновлять складские запасы. Мы используем CQRS и Message Bus.

using Wolverine;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.DependencyInjection;

// --- Модели ---
public record PlaceOrder(int OrderId, string Product, int Quantity, string CustomerEmail);
public record OrderPlaced(int OrderId, string Product, int Quantity);
public record SendNotification(string Email, string Message);
public record UpdateInventory(string Product, int Quantity);

// --- Обработчики ---
public class PlaceOrderHandler
{
    private readonly IMessageBus _bus;

    public PlaceOrderHandler(IMessageBus bus)
    {
        _bus = bus;
    }

    public async Task Handle(PlaceOrder command)
    {
        // 1. Сохраняем заказ в БД (опущено для краткости)
        Console.WriteLine($"Order {command.OrderId} placed for {command.Product} x{command.Quantity}");

        // 2. Публикуем события
        await _bus.PublishAsync(new OrderPlaced(command.OrderId, command.Product, command.Quantity));
        await _bus.PublishAsync(new SendNotification(command.CustomerEmail, $"Order {command.OrderId} confirmed!"));
        await _bus.PublishAsync(new UpdateInventory(command.Product, command.Quantity));
    }
}

public class NotificationHandler
{
    public void Handle(SendNotification notification)
    {
        // Симуляция отправки email
        Console.WriteLine($"Email sent to {notification.Email}: {notification.Message}");
    }
}

public class InventoryHandler
{
    public void Handle(UpdateInventory update)
    {
        // Симуляция обновления склада
        Console.WriteLine($"Inventory updated: {update.Product} reduced by {update.Quantity}");
    }
}

// --- Программа ---
public class Program
{
    public static async Task Main(string[] args)
    {
        using var host = await Host.CreateDefaultBuilder(args)
            .UseWolverine(opts =>
            {
                // Настройка: все сообщения обрабатываются локально
                opts.PublishAllMessages().Locally();
                
                // Включаем автоматические транзакции (если бы была БД)
                opts.Policies.AutoApplyTransactions();
            })
            .StartAsync();

        var bus = host.Services.GetRequiredService();

        // Отправляем команду на размещение заказа
        await bus.SendAsync(new PlaceOrder(1, "Laptop", 2, "customer@example.com"));

        Console.WriteLine("Order processing completed.");
        await host.StopAsync();
    }
}

В этом примере команда PlaceOrder обрабатывается, и после успешного сохранения заказа публикуются три события: OrderPlaced, SendNotification и UpdateInventory. Каждое из этих событий обрабатывается своим обработчиком. Это демонстрирует принцип CQRS и асинхронное взаимодействие через Message Bus.

6. Советы по производительности и типичные ошибки

Советы по производительности:

Используйте локальные очереди для быстрых операций: Если обработчик выполняется быстро (менее 100 мс), используйте EnqueueAsync или InvokeAsync вместо отправки во внешнюю очередь, чтобы избежать накладных расходов на сериализацию и сеть.
Настройте параллелизм: Wolverine позволяет настроить количество параллельно обрабатываемых сообщений для каждого слушателя. Увеличьте это значение для обработчиков, которые выполняют I/O-bound операции.
Используйте пул соединений: При работе с внешними очередями (RabbitMQ, Azure Service Bus) убедитесь, что вы используете пул соединений, чтобы избежать создания нового соединения для каждого сообщения.
Минимизируйте размер сообщений: Передавайте только необходимые данные. Избегайте передачи больших объектов или файлов внутри сообщений.
Используйте ValueTask в обработчиках: Если ваш обработчик может завершиться синхронно, используйте ValueTask вместо Task, чтобы уменьшить количество аллокаций.