Как использовать ID сообщения для дедупликации?

Дедупликация сообщений — это процесс обнаружения и отбрасывания повторяющихся сообщений, чтобы одна и та же операция не выполнялась несколько раз. В распределённых системах, где сообщения могут быть доставлены повторно из-за сбоев сети, таймаутов или ретраев, это необходимо для поддержания целостности данных и предотвращения ошибок, таких как двойное списание средств.

Как работает дедупликация на основе ID

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

Типичная реализация

Система, выполняющая обработку, поддерживает кэш или таблицу уже обработанных ID. Процесс выглядит так:

1. При получении сообщения извлекается его ID.
2. Проверяется, существует ли этот ID в хранилище обработанных ID.
3. Если ID найден, сообщение считается дубликатом и отбрасывается (или подтверждается без повторной обработки).
4. Если ID не найден, сообщение обрабатывается, а его ID сохраняется в хранилище.

Хранилище должно иметь TTL (время жизни), чтобы старые ID автоматически удалялись и не занимали память бесконечно. Для этого отлично подходят Redis или аналогичные key-value хранилища.

Пример кода на Python

Вот упрощённый пример обработчика сообщений с использованием Redis для дедупликации:

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)
TTL_SECONDS = 3600  # Храним ID 1 час

def process_message_with_deduplication(message):
    """Обрабатывает сообщение, пропуская дубликаты."""
    message_id = message.get('id')
    if not message_id:
        raise ValueError("Сообщение не содержит ID")
    
    # Пытаемся добавить ID в Redis. Ключ будет существовать только если его ещё нет.
    # Команда SET с опциями NX и EX делает это атомарно.
    is_new = redis_client.set(message_id, "processed", ex=TTL_SECONDS, nx=True)
    
    if not is_new:
        print(f"Сообщение с ID {message_id} уже обработано, пропускаем.")
        return  # Это дубликат
    
    # Основная логика обработки сообщения
    print(f"Обрабатываем новое сообщение: {json.dumps(message)}")
    # ... ваша бизнес-логика здесь ...
    # Например, обновление баланса пользователя
    # Важно: эта логика должна быть идемпотентной сама по себе на случай сбоя после сохранения ID.

# Пример вызова
msg = {"id": "550e8400-e29b-41d4-a716-446655440000", "user": "alice", "amount": 100}
process_message_with_deduplication(msg)

Где применяется

Этот подход широко используется в:

• Очередях сообщений (RabbitMQ, Kafka) для гарантированной доставки.
• Платежных системах для предотвращения двойного списания.
• Синхронизации данных между микросервисами.
• Обработке событий от внешних API, которые могут повторно отправлять запросы.

Вывод: Использование ID сообщения для дедупликации — это фундаментальный паттерн для построения надёжных распределённых систем. Его стоит применять везде, где критически важна идемпотентность операций и возможны повторные доставки сообщений.