Где хранить историю отправки сообщений в RabbitMQ?

Архитектура сохранения истории сообщений в RabbitMQ

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

1. Основные принципы и термины

Ключевые термины в контексте RabbitMQ:

• Producer (публикует сообщения в очередь)
• Exchange (маршрутизирует сообщения)
• Queue (хранит сообщения для потребителей)
• Consumer (обрабатывает сообщения из очереди)
• Message Persistence (сохранение сообщения на диске RabbitMQ)

RabbitMQ по умолчанию не хранит историю отправленных сообщений после их обработки потребителем. После успешной обработки сообщение удаляется из очереди. Поэтому для сохранения истории требуется дополнительная архитектура.

2. Основные подходы к сохранению истории

2.1. Логирование на стороне Producer (наиболее распространённый)

Producer сохраняет каждое отправленное сообщение в локальное хранилище (базу данных, файловый лог) перед отправкой в RabbitMQ.

// Пример класса Producer с сохранением истории
class MessageProducer {
    private $repository;
    private $rabbitMQConnection;

    public function sendMessage(array $payload, string $queueName): void {
        // 1. Сохраняем в историю перед отправкой
        $messageId = $this->repository->saveMessageHistory([
            'payload' => $payload,
            'queue' => $queueName,
            'status' => 'pending',
            'created_at' => new DateTime()
        ]);

        // 2. Отправляем в RabbitMQ
        try {
            $this->rabbitMQConnection->publish($queueName, [
                'body' => $payload,
                'message_id' => $messageId // передаём ID для отслеживания
            ]);
            
            // 3. Обновляем статус в истории
            $this->repository->updateStatus($messageId, 'sent');
        } catch (Exception $e) {
            $this->repository->updateStatus($messageId, 'failed', $e->getMessage());
        }
    }
}

Преимущества:

• Полная история всех отправленных сообщений
• Возможность отслеживания статуса (sent, failed, processed)
• Лёгкий аудит и диагностика проблем

Недостатки:

• Двойная запись (в БД и RabbitMQ)
• Возможные расхождения между историей и фактической отправкой

2.2. Использование подтверждений (Acknowledgements) и повторных отправок

RabbitMQ поддерживает механизм ACK/NACK, который позволяет отслеживать обработку сообщений.

// Пример Consumer с подтверждением и сохранением истории обработки
class MessageConsumer {
    public function consume(AMQPMessage $message): void {
        $messageId = $message->get('message_id');
        
        try {
            // Обработка сообщения
            $this->processMessage($message->body);
            
            // Сохраняем в историю УСПЕШНОЙ обработки
            $this->repository->saveConsumerHistory([
                'message_id' => $messageId,
                'status' => 'processed',
                'processed_at' => new DateTime()
            ]);
            
            // Подтверждаем обработку (ACK)
            $message->ack();
        } catch (Exception $e) {
            // Сохраняем в историю ОШИБКИ обработки
            $this->repository->saveConsumerHistory([
                'message_id' => $messageId,
                'status' => 'error',
                'error' => $e->getMessage()
            ]);
            
            // Отказываемся от сообщения (NACK) с требованием повторной отправки
            $message->nack(true);
        }
    }
}

2.3. Архитектура "двойной записи" через шину событий (Event Bus)

Применяется в сложных системах, где требуется полный аудит всех событий.

Producer -> [Сохранение в БД истории] -> RabbitMQ -> Consumer -> [Сохранение в БД обработки]

3. Технические реализации хранилища истории

Выбор хранилища зависит от объёма данных и требований к поиску:

• SQL базы данных (MySQL, PostgreSQL) - для структурированной истории с сложными запросами
• NoSQL (MongoDB, Elasticsearch) - для больших объёмов и полнотекстового поиска в payload
• Файловые логи (JSON, CSV) - для простых случаев и дебага
• Специализированные системы (Kafka с длительным retention) - для high-load систем

4. Практические рекомендации для PHP Backend

1. Используйте correlation_id - передавайте уникальный идентификатор сообщения через headers RabbitMQ для связывания отправки и обработки.

// При отправке сообщения
$channel->basic_publish(
    new AMQPMessage($body, [
        'correlation_id' => uniqid('msg_', true),
        'app_id' => 'my_backend_app',
        'headers' => ['source' => 'user_service']
    ]),
    $exchange
);

1. Реализуйте отдельный сервис аудита - который централизованно собирает историю из всех Producers и Consumers.

2. Учитывайте требования GDPR и безопасности - некоторые payload могут содержать персональные данные, их история должна храниться с шифрованием или ограниченным доступом.

3. Для высоконагруженных систем используйте sampling - сохраняйте не каждое сообщение, а только 1% или по определённым правилам (например, только ошибки).

5. Полная архитектурная схема

[Producer Service]
       ↓ (сохраняет в Audit DB)
[RabbitMQ Exchange]
       ↓ (маршрутизация)
[RabbitMQ Queue]
       ↓
[Consumer Service]
       ↓ (сохраняет в Audit DB)
[Audit Service] ← агрегирует данные из всех источников

Итог: История отправки сообщений должна храниться вне RabbitMQ - в отдельном хранилище (базе данных, лог-файлах), куда данные записываются либо Producer перед отправкой, либо Consumer после обработки, либо отдельным сервисом аудита, отслеживающим весь поток сообщений. RabbitMQ сам является транспортным механизмом, а не системой хранения истории.