В каких сценариях использовал RabbitMQ?

Опыт использования RabbitMQ в C# Backend-разработке

За 10+ лет работы с распределенными системами я использовал RabbitMQ в различных сценариях, где требовалась надежная асинхронная коммуникация между микросервисами и компонентами системы. Вот ключевые кейсы:

1. Асинхронная обработка длительных операций

Использовал для выноса ресурсоемких задач из синхронного HTTP-цикла:

• Генерация отчетов и документов (PDF, Excel)
• Обработка медиафайлов (конвертация видео, оптимизация изображений)
• Интеграция с внешними API с непредсказуемым временем ответа

// Пример публикации задачи на обработку
public async Task PublishReportGenerationTask(ReportRequest request)
{
    using var channel = _connection.CreateModel();
    var properties = channel.CreateBasicProperties();
    properties.Persistent = true; // Сохраняем сообщение при перезагрузке
    properties.Headers = new Dictionary<string, object>
    {
        ["ReportType"] = request.Type,
        ["Priority"] = "high"
    };
    
    var body = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(request));
    channel.BasicPublish(
        exchange: "reports.exchange",
        routingKey: "report.generate",
        basicProperties: properties,
        body: body);
}

2. Микросервисная архитектура и Event-Driven Design

• Декапляция сервисов: Сервис уведомлений не знает о сервисе заказов
• Синхронизация данных между БД разных сервисов через события
• Реализация Saga Pattern для управления распределенными транзакциями

// Конфигурация обмена для событий домена
public void ConfigureEventBus(IServiceCollection services)
{
    services.AddSingleton<IRabbitMQPersistentConnection>(sp => { /* ... */ });
    
    services.AddSingleton<IEventBus, EventBusRabbitMQ>(sp =>
    {
        var subscriptionClientName = "ordering-service";
        return new EventBusRabbitMQ(
            connection: sp.GetRequiredService<IRabbitMQPersistentConnection>(),
            logger: sp.GetRequiredService<ILogger<EventBusRabbitMQ>>(),
            subscriptionClientName: subscriptionClientName,
            retryCount: 5);
    });
}

3. Балансировка нагрузки между воркерами

Настройка Competing Consumers Pattern для горизонтального масштабирования:

// Конфигурация очереди с несколькими потребителями
public void SetupWorkQueue()
{
    var channel = _connection.CreateModel();
    
    // Декларация очереди с параметрами надежности
    channel.QueueDeclare(
        queue: "image-processing.queue",
        durable: true,          // Переживет перезагрузку RabbitMQ
        exclusive: false,
        autoDelete: false,
        arguments: new Dictionary<string, object>
        {
            ["x-max-priority"] = 10, // Поддержка приоритетов
            ["x-dead-letter-exchange"] = "dead.letters.exchange"
        });
    
    // Quality of Service - контроль нагрузки
    channel.BasicQos(
        prefetchSize: 0,
        prefetchCount: 1,       // По 1 сообщению на воркера
        global: false);
    
    // Несколько потребителей
    for (int i = 0; i < 5; i++)
    {
        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += async (model, ea) =>
        {
            // Обработка сообщения
            await ProcessImageAsync(ea.Body.ToArray());
            channel.BasicAck(ea.DeliveryTag, multiple: false);
        };
        channel.BasicConsume("image-processing.queue", autoAck: false, consumer);
    }
}

4. Буферизация и защита от пиковых нагрузок

• Rate limiting: Накопление запросов при превышении лимитов внешних API
• Circuit Breaker Pattern: Временное хранение сообщений при недоступности сервиса-получателя
• Очередь ретраев для повторной обработки неудачных операций

5. Распределение событий через Topic Exchange

Сложные маршрутизации для систем нотификации:

// Подписка на определенные типы событий
public void SubscribeToNotifications()
{
    channel.ExchangeDeclare("notifications.topic", ExchangeType.Topic);
    var queueName = channel.QueueDeclare().QueueName;
    
    // Подписка на все email-уведомления для пользователей EU
    channel.QueueBind(
        queue: queueName,
        exchange: "notifications.topic",
        routingKey: "email.eu.*");
    
    // Подписка на SMS для высокоприоритетных событий
    channel.QueueBind(
        queue: queueName,
        exchange: "notifications.topic",
        routingKey: "sms.high.*");
}

6. Мониторинг и dead letter queues (DLX)

Настройка обработки проблемных сообщений:

public void ConfigureDeadLetterHandling()
{
    // Основная очередь с указанием DLX
    var args = new Dictionary<string, object>
    {
        {"x-dead-letter-exchange", "dlx.exchange"},
        {"x-dead-letter-routing-key", "failed.payments"},
        {"x-message-ttl", 60000} // TTL 60 секунд
    };
    
    channel.QueueDeclare("payment.processing.queue", durable: true, false, false, args);
    
    // Dead Letter Queue для анализа проблем
    channel.QueueDeclare("failed.payments.queue", durable: true, false, false);
    channel.ExchangeDeclare("dlx.exchange", ExchangeType.Direct);
    channel.QueueBind("failed.payments.queue", "dlx.exchange", "failed.payments");
}

Ключевые преимущества RabbitMQ в этих сценариях:

• Надежность: Подтверждения доставки (ack/nack), persistent messages
• Гибкость маршрутизации: 4 типа exchange (direct, fanout, topic, headers)
• Масштабируемость: Clustering, mirrored queues, shovel/federation
• Управление потоком: QoS, priority queues, TTL
• Мониторинг: Management UI, Prometheus metrics, tracing

Проблемы и решения:

1. Потеря сообщений → Включение publisher confirms и persistent messages
2. Дублирование сообщений → Идемпотентная обработка на стороне потребителя
3. Накопление очереди → Настройка TTL, dead letter exchanges, мониторинг глубины очереди
4. Сложность отладки → Использование correlation id, трассировка распределенных транзакций

RabbitMQ стал неотъемлемой частью архитектуры в проектах с высокой нагрузкой (10K+ сообщений/сек), где требовалась гарантированная доставка и гибкая маршрутизация сообщений между десятками микросервисов.