Event-Driven архитектуры в C# - сравнение брокеров и выбор решения

// Пример Azure Function с триггером Service Bus
public static class OrderProcessor
{
    [FunctionName("ProcessOrder")]
    public static async Task Run(
        [ServiceBusTrigger("orders", Connection = "ServiceBusConnection")] string message,
        ILogger log)
    {
        log.LogInformation($"Получено сообщение: {message}");
        
        // Обработка заказа
        var order = JsonSerializer.Deserialize<Order>(message);
        await ProcessOrderAsync(order);
    }
}

// Отправка сообщения с метаданными для фильтрации
var message = new ServiceBusMessage(Encoding.UTF8.GetBytes(orderJson));
message.ApplicationProperties.Add("Region", "Europe");
message.ApplicationProperties.Add("OrderType", "Premium");
message.ApplicationProperties.Add("Amount", 5000);
 
await sender.SendMessageAsync(message);
 
// Создание подписки с SQL-фильтром
var adminClient = new ServiceBusAdministrationClient(connectionString);
var subscriptionOptions = new CreateSubscriptionOptions("orders", "europe-premium-orders");
var rule = new CreateRuleOptions("premium-europe-filter", 
    new SqlRuleFilter("Region = 'Europe' AND OrderType = 'Premium' AND Amount > 1000"));
 
await adminClient.CreateSubscriptionAsync(subscriptionOptions);
await adminClient.CreateRuleAsync("orders", "europe-premium-orders", rule);

// Отправка сообщения с привязкой к сессии
var message = new ServiceBusMessage(Encoding.UTF8.GetBytes(orderJson));
message.SessionId = "customer-123"; // Все сообщения с одинаковым SessionId 
                                    // будут обработаны последовательно
await sender.SendMessageAsync(message);
 
// Получение сообщений конкретной сессии
var sessionReceiver = await client.AcceptSessionAsync("orders", "customer-123");
var messages = await sessionReceiver.ReceiveMessagesAsync(10);

// Настройка процессора с оптимальными параметрами
var options = new ServiceBusProcessorOptions
{
    MaxConcurrentCalls = 10, // Максимальное количество параллельных вызовов
    AutoCompleteMessages = false, // Ручное подтверждение для надежности
    PrefetchCount = 100 // Предварительная загрузка сообщений для 
                        // оптимизации производительности
};
 
var processor = client.CreateProcessor("orders", options);
processor.ProcessMessageAsync += MessageHandler;
processor.ProcessErrorAsync += ErrorHandler;
 
await processor.StartProcessingAsync();

// Получение сообщений из очереди мертвых писем
var dlqReceiver = client.CreateReceiver("orders", 
    new ServiceBusReceiverOptions { SubQueue = SubQueue.DeadLetter });
 
var deadLetterMessages = await dlqReceiver.ReceiveMessagesAsync(10);
 
foreach (var message in deadLetterMessages)
{
    Console.WriteLine($"Проблемное сообщение: {message.Body}, " +
                    $"причина: {message.DeadLetterReason}, " +
                    $"описание: {message.DeadLetterErrorDescription}");
    
    // Анализ и исправление проблемы
    // ...
    
    // Повторная отправка в основную очередь
    await sender.SendMessageAsync(new ServiceBusMessage(message.Body));
    
    // Подтверждение обработки мертвого сообщения
    await dlqReceiver.CompleteMessageAsync(message);
}

public class OrderSaga : MassTransitStateMachine<OrderSagaState>
{
    public OrderSaga()
    {
        InstanceState(x => x.CurrentState);
        
        // Инициируем сагу при создании заказа
        Initially(
            When(OrderSubmitted)
                .Then(context => InitializeOrder(context))
                .TransitionTo(AwaitingValidation)
                .Publish(context => new ValidateOrderCommand { OrderId = context.Instance.OrderId }));
        
        // Ожидаем валидацию
        During(AwaitingValidation,
            When(OrderValidated)
                .TransitionTo(AwaitingPayment)
                .Publish(context => new ProcessPaymentCommand { OrderId = context.Instance.OrderId }),
            When(OrderValidationFailed)
                .TransitionTo(ValidationFailed)
                .Publish(context => new CancelOrderCommand { OrderId = context.Instance.OrderId }));
                
        // Ожидаем оплату
        During(AwaitingPayment,
            When(PaymentSucceeded)
                .TransitionTo(PaymentCompleted)
                .Publish(context => new ShipOrderCommand { OrderId = context.Instance.OrderId }),
            When(PaymentFailed)
                .TransitionTo(PaymentRejected)
                .Publish(context => new CancelOrderCommand { OrderId = context.Instance.OrderId }));
                
        // Остальные состояния и переходы...
    }
}

// Обработчик команды (запись)
public async Task Handle(CreateOrderCommand command, CancellationToken token)
{
    var order = new Order(command.OrderId, command.CustomerId);
    order.AddItems(command.Items);
    
    // Сохраняем события, а не состояние
    var events = order.GetUncommittedEvents();
    await _eventStore.SaveEventsAsync(command.OrderId, events);
    
    // Публикуем события для обновления read-моделей
    foreach (var @event in events)
    {
        await _eventBus.PublishAsync(@event);
    }
}
 
// Обработчик запроса (чтение)
public async Task<OrderDto> Handle(GetOrderQuery query, CancellationToken token)
{
    // Читаем из оптимизированной read-модели
    return await _orderReadRepository.GetByIdAsync(query.OrderId);
}

// Пример с Polly
var retryPolicy = Policy
    .Handle<TransientException>()
    .WaitAndRetryAsync(
        retryCount: 5,
        sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
        onRetry: (ex, time, attempt, ctx) => 
            _logger.LogWarning(ex, "Попытка {Attempt} обработки сообщения после {Time}", attempt, time));
            
await retryPolicy.ExecuteAsync(() => ProcessMessageAsync(message));

public class OrderService : IOrderService
{
    private readonly IKafkaProducer _kafkaProducer;
    private readonly IRabbitMqPublisher _rabbitPublisher;
    private readonly INatsClient _natsClient;
    private readonly IServiceBusSender _serviceBusSender;
    private readonly IOrderRepository _orderRepository;
    private readonly ILogger<OrderService> _logger;
 
    public OrderService(
        IKafkaProducer kafkaProducer,
        IRabbitMqPublisher rabbitPublisher,
        INatsClient natsClient,
        IServiceBusSender serviceBusSender,
        IOrderRepository orderRepository,
        ILogger<OrderService> logger)
    {
        _kafkaProducer = kafkaProducer;
        _rabbitPublisher = rabbitPublisher;
        _natsClient = natsClient;
        _serviceBusSender = serviceBusSender;
        _orderRepository = orderRepository;
        _logger = logger;
    }
 
    public async Task<OrderResult> CreateOrderAsync(OrderRequest request)
    {
        // 1. Быстрая проверка доступности товара через NATS (запрос-ответ)
        var inventoryCheckResult = await _natsClient.RequestAsync<InventoryCheckRequest, InventoryCheckResponse>(
            "inventory.check",
            new InventoryCheckRequest { ProductIds = request.Items.Select(i => i.ProductId).ToList() });
 
        if (!inventoryCheckResult.AllAvailable)
        {
            return OrderResult.Failed("Некоторые товары недоступны");
        }
 
        // 2. Создание заказа в базе данных
        var order = new Order
        {
            Id = Guid.NewGuid(),
            CustomerId = request.CustomerId,
            Status = OrderStatus.Created,
            Items = request.Items.Select(i => new OrderItem
            {
                ProductId = i.ProductId,
                Quantity = i.Quantity,
                Price = i.Price
            }).ToList(),
            TotalAmount = request.Items.Sum(i => i.Price * i.Quantity),
            CreatedAt = DateTime.UtcNow
        };
 
        await _orderRepository.CreateAsync(order);
 
        // 3. Отправка команды резервирования товаров через RabbitMQ
        await _rabbitPublisher.PublishAsync(
            "inventory.commands",
            "reserve",
            new ReserveInventoryCommand
            {
                OrderId = order.Id,
                Items = order.Items.Select(i => new ReserveItem
                {
                    ProductId = i.ProductId,
                    Quantity = i.Quantity
                }).ToList()
            });
 
        // 4. Публикация события создания заказа в Kafka для аналитики и аудита
        await _kafkaProducer.ProduceAsync(
            "order-events",
            order.Id.ToString(),
            new OrderCreatedEvent
            {
                OrderId = order.Id,
                CustomerId = order.CustomerId,
                Items = order.Items.Select(i => new OrderItemDto
                {
                    ProductId = i.ProductId,
                    Quantity = i.Quantity,
                    Price = i.Price
                }).ToList(),
                TotalAmount = order.TotalAmount,
                CreatedAt = order.CreatedAt
            });
 
        // 5. Отправка уведомления через Azure Service Bus
        await _serviceBusSender.SendMessageAsync(
            new CustomerNotificationMessage
            {
                CustomerId = order.CustomerId,
                Subject = "Заказ создан",
                Body = $"Ваш заказ #{order.Id} на сумму {order.TotalAmount} успешно создан.",
                Type = NotificationType.OrderCreated,
                OrderId = order.Id
            });
 
        return OrderResult.Success(order.Id);
    }
}

public class MessageBrokerBenchmark
{
    private const int MessageCount = 1000;
    private readonly IKafkaProducer _kafkaProducer;
    private readonly IRabbitMqPublisher _rabbitPublisher;
    private readonly INatsClient _natsClient;
    private readonly IServiceBusSender _serviceBusSender;
 
    public async Task RunBenchmarkAsync()
    {
        var message = GenerateTestMessage();
        
        // Разогрев системы
        await SendWarmupMessagesAsync();
        
        // Kafka benchmark
        var kafkaStopwatch = Stopwatch.StartNew();
        for (int i = 0; i < MessageCount; i++)
        {
            await _kafkaProducer.ProduceAsync("benchmark", Guid.NewGuid().ToString(), message);
        }
        kafkaStopwatch.Stop();
        
        // RabbitMQ benchmark
        var rabbitStopwatch = Stopwatch.StartNew();
        for (int i = 0; i < MessageCount; i++)
        {
            await _rabbitPublisher.PublishAsync("benchmark", "test", message);
        }
        rabbitStopwatch.Stop();
        
        // NATS benchmark
        var natsStopwatch = Stopwatch.StartNew();
        for (int i = 0; i < MessageCount; i++)
        {
            await _natsClient.PublishAsync("benchmark.test", message);
        }
        natsStopwatch.Stop();
        
        // Azure Service Bus benchmark
        var serviceBusStopwatch = Stopwatch.StartNew();
        for (int i = 0; i < MessageCount; i++)
        {
            await _serviceBusSender.SendMessageAsync(message);
        }
        serviceBusStopwatch.Stop();
        
        // Вывод результатов
        Console.WriteLine($"Kafka: {kafkaStopwatch.ElapsedMilliseconds} ms ({MessageCount * 1000 / kafkaStopwatch.ElapsedMilliseconds} msg/sec)");
        Console.WriteLine($"RabbitMQ: {rabbitStopwatch.ElapsedMilliseconds} ms ({MessageCount * 1000 / rabbitStopwatch.ElapsedMilliseconds} msg/sec)");
        Console.WriteLine($"NATS: {natsStopwatch.ElapsedMilliseconds} ms ({MessageCount * 1000 / natsStopwatch.ElapsedMilliseconds} msg/sec)");
        Console.WriteLine($"Azure Service Bus: {serviceBusStopwatch.ElapsedMilliseconds} ms ({MessageCount * 1000 / serviceBusStopwatch.ElapsedMilliseconds} msg/sec)");
    }
}

public async Task CreateOrderWithOutboxAsync(OrderRequest request)
{
    using var transaction = await _dbContext.Database.BeginTransactionAsync();
    
    try
    {
        // Создание заказа
        var order = new Order { /* ... */ };
        _dbContext.Orders.Add(order);
        
        // Создание записи в outbox
        var outboxMessage = new OutboxMessage
        {
            Id = Guid.NewGuid(),
            Topic = "order-events",
            Key = order.Id.ToString(),
            Payload = JsonSerializer.Serialize(new OrderCreatedEvent { /* ... */ }),
            CreatedAt = DateTime.UtcNow,
            ProcessedAt = null,
            MessageType = "OrderCreatedEvent",
            BrokerType = "Kafka" // Указываем, через какой брокер отправлять
        };
        
        _dbContext.OutboxMessages.Add(outboxMessage);
        await _dbContext.SaveChangesAsync();
        
        await transaction.CommitAsync();
    }
    catch
    {
        await transaction.RollbackAsync();
        throw;
    }
}