Как выполнишь batch обработку сообщений в RabbitMQ?

Стратегии Batch Processing в RabbitMQ

Batch processing в RabbitMQ — это подход, позволяющий обрабатывать несколько сообщений как единую группу, что повышает эффективность и снижает нагрузку на систему. Это особенно полезно при высокой нагрузке или при необходимости агрегирования данных.

Основные методы реализации

1. Prefetch Count (QoS) — Базовая настройка потребителя

RabbitMQ позволяет потребителю заранее запрашивать несколько сообщений в свой локальный буфер через параметр PrefetchCount. Это не строгий batch, но позволяет уменьшить частоту сетевых запросов.

using RabbitMQ.Client;

var factory = new ConnectionFactory() { HostName = "localhost" };
var connection = factory.CreateConnection();
var channel = connection.CreateModel();

// Устанавливаем PrefetchCount для канала
channel.BasicQos(prefetchSize: 0, prefetchCount: 50, global: false);

var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) => {
    // Обработка одного сообщения, но из локального буфера
    var body = ea.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    // Логика обработки...
};

channel.BasicConsume(queue: "myQueue", autoAck: false, consumer: consumer);

2. Агрегирование сообщений на стороне потребителя

Это наиболее распространенный подход: потребитель накапливает сообщения в памяти до достижения определенного лимита (по количеству или времени), затем обрабатывает их как batch.

public class BatchConsumer
{
    private readonly List<byte[]> _messageBatch = new List<byte[]>();
    private readonly int _batchSize = 100;
    private readonly Timer _batchTimer;
    private readonly IModel _channel;

    public BatchConsumer(IModel channel)
    {
        _channel = channel;
        _batchTimer = new Timer(FlushBatch, null, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(5));
    }

    public void AddMessage(byte[] body)
    {
        _messageBatch.Add(body);
        if (_messageBatch.Count >= _batchSize)
        {
            ProcessBatch();
        }
    }

    private void ProcessBatch()
    {
        if (_messageBatch.Count == 0) return;

        // Пример обработки batch: преобразование всех сообщений
        var messages = _messageBatch.Select(body => Encoding.UTF8.GetString(body)).ToList();
        
        // Логика batch обработки (например, сохранение в БД bulk insert)
        SaveToDatabase(messages);
        
        // Подтверждение обработки всех сообщений в batch
        foreach (var deliveryTag in GetDeliveryTags()) // deliveryTags нужно хранить отдельно
        {
            _channel.BasicAck(deliveryTag, multiple: true);
        }
        
        _messageBatch.Clear();
    }

    private void FlushBatch(object state) => ProcessBatch();
}

3. Batch подтверждение (Multi Ack)

RabbitMQ поддерживает подтверждение нескольких сообщений одним запросом через BasicAck с параметром multiple: true. Это эффективно снижает сетевой трафик.

// После обработки batch
var lastDeliveryTag = deliveryTags.Last(); // deliveryTags собраны из сообщений
channel.BasicAck(deliveryTag: lastDeliveryTag, multiple: true);

Ключевые рекомендации и подводные камни

• Управление памятью: Batch обработка требует аккуратного контроля объема памяти, особенно при больших размерах сообщений. Рекомендуется устанавливать пределы по количеству (batchSize) и времени (batchTimeout).
• Отказоустойчивость: Если обработка batch завершилась ошибкой, необходимо иметь стратегию повторной обработки или восстановления. Можно использовать transactional batches (не рекомендуются из-за производительности) или сохранять batch в промежуточное хранилище.
• Подтверждение сообщений: При использовании multiple: true в BasicAck все сообщения до указанного deliveryTag будут подтверждены. Необходимо строго отслеживать порядок получения сообщений.
• Производительность сети: Batch обработка значительно снижает частоту сетевых взаимодействий, но может увеличить latency для отдельных сообщений.

Альтернативный подход: Batch отправка (Producer Side)

RabbitMQ не поддерживает нативный batch отправку, но можно агрегировать сообщения перед отправкой:

public class BatchPublisher
{
    private readonly List<string> _pendingMessages = new List<string>();
    private readonly int _batchLimit = 50;

    public void PublishBatch(IModel channel, string exchange, string routingKey)
    {
        var batchProperties = channel.CreateBasicProperties();
        batchProperties.Persistent = true;

        foreach (var message in _pendingMessages)
        {
            channel.BasicPublish(exchange: exchange,
                                routingKey: routingKey,
                                basicProperties: batchProperties,
                                body: Encoding.UTF8.GetBytes(message));
        }
        _pendingMessages.Clear();
    }
}

Выбор стратегии в зависимости от требований

• Высокая пропускная способность: Используйте агрегирование на стороне потребителя с большим PrefetchCount и batch подтверждением.
• Минимальный latency: Уменьшите размер batch или используйте только PrefetchCount без агрегирования.
• Гарантия обработки: Включите механизм повторных попыток и мониторинг размера batch.

Batch обработка в RabbitMQ требует баланса между эффективностью, надежностью и timeliness сообщений. Правильная реализация может увеличить пропускную способность системы в 5-10 раз при высокой нагрузке.