> Как обрабатывать дубликаты сообщений при асинхронной записи из очереди в базу (Go)

Короткий ответ

Дубликаты сообщений при асинхронной записи решаются через идемпотентность: уникальный ключ сообщения (message ID) и проверка его наличия перед вставкой. В PostgreSQL - уникальный индекс + INSERT ... ON CONFLICT DO NOTHING, в Redis - SETNX, в Cassandra - INSERT IF NOT EXISTS. Дополнительно нужен механизм очистки старых ключей (TTL или периодическая чистка), чтобы таблица не росла бесконечно. Важно учитывать, что проверка и вставка должны быть атомарными, иначе гонка останется.

Подробное объяснение

Проблема возникает из-за того, что consumer может получить одно и то же сообщение несколько раз: при ретраях, после ребалансировки партиций, при at-least-once доставке. Если писать в базу без проверки, получим дубликаты строк.

Основные подходы:

1. Уникальный ключ из данных сообщения - если бизнес-логика позволяет, используем естественный ключ (например, order_id). Тогда дубликат физически не вставится.

2. Отдельная таблица processed_messages - храним message_id, который уже обработан. Перед записью проверяем наличие. Минус: лишний запрос, но при правильном индексировании это быстро.

3. Идемпотентная запись через unique constraint - самый надёжный вариант. Создаём уникальный индекс на поле message_id в целевой таблице, а вставку делаем с ON CONFLICT DO NOTHING (PostgreSQL) или INSERT IGNORE (MySQL). Это атомарно и не требует отдельной проверки.

4. Redis как фильтр - SETNX message_id 1 EX 86400 перед записью. Если вернул 1 - сообщение новое, пишем в базу. Если 0 - пропускаем. Минус: Redis может потерять данные, поэтому нужен fallback.

5. Дедупликация на уровне БД через транзакцию - SELECT ... FOR UPDATE на ключе, затем вставка. Работает, но медленнее и блокирует строки.

Для Go важно: если используем несколько воркеров, нужно, чтобы проверка и вставка были в одной транзакции или использовали атомарную операцию БД. Иначе два воркера могут одновременно прочитать, что ключа нет, и оба вставят.

На практике

Для высоконагруженных систем рекомендую комбинацию:

• Уникальный индекс на message_id в целевой таблице - это гарантия на уровне БД.
• INSERT ... ON CONFLICT DO NOTHING - не требует предварительного SELECT.
• Если сообщений много и таблица большая, можно партиционировать по времени и удалять старые партиции.
• Для Redis-фильтра используйте TTL, равный максимальному времени ретрая (например, 24 часа). Это защитит от повторной обработки старых сообщений, но не будет хранить ключи вечно.

Если пишем в несколько таблиц (например, заказ и его позиции), нужна транзакция. В PostgreSQL можно использовать ON CONFLICT внутри транзакции - это безопасно.

Пример кода

GO
// PostgreSQL + pgx
func insertOrder(ctx context.Context, tx pgx.Tx, msg Message) error {
    _, err := tx.Exec(ctx, `
        INSERT INTO orders (id, message_id, payload, created_at)
        VALUES ($1, $2, $3, now())
        ON CONFLICT (message_id) DO NOTHING
    `, msg.OrderID, msg.MessageID, msg.Payload)
    return err
}

// Redis фильтр + запись в БД
func processMessage(ctx context.Context, msg Message) error {
    ok, err := rdb.SetNX(ctx, "msg:"+msg.MessageID, "1", 24*time.Hour).Result()
    if err != nil {
        return err
    }
    if !ok {
        log.Printf("duplicate message %s", msg.MessageID)
        return nil
    }
    // запись в БД
    _, err = db.Exec(ctx, `
        INSERT INTO orders (id, message_id, payload)
        VALUES ($1, $2, $3)
    `, msg.OrderID, msg.MessageID, msg.Payload)
    return err
}

Как отвечать на собеседовании

Начните с формулировки проблемы: at-least-once доставка и ретраи приводят к дубликатам. Затем предложите идемпотентность как основной принцип. Упомяните, что надёжнее всего - уникальный индекс в БД, а Redis - как быстрый фильтр для снижения нагрузки. Обязательно скажите про атомарность: проверка и вставка должны быть одной операцией. Если спросят про производительность - отметьте, что ON CONFLICT быстрее, чем SELECT + INSERT, потому что не требует round-trip. Приведите пример с Go и pgx.

Что проверяет интервьюер

• Понимание моделей доставки (at-least-once, exactly-once) и их следствий.
• Знание идемпотентности и способов её реализации.
• Умение выбирать между БД и внешними хранилищами для дедупликации.
• Понимание гонок и необходимости атомарных операций.
• Практический опыт с конкретными СУБД и драйверами.

Типичные ошибки

• Предложение делать SELECT перед INSERT без транзакции - это не защищает от гонки.
• Игнорирование TTL для Redis-ключей - таблица или Redis растут бесконечно.
• Использование только Redis без fallback - при потере данных Redis дубликаты пройдут.
• Неучёт того, что ON CONFLICT DO NOTHING не возвращает, была ли вставка - если нужно понять, новое сообщение или дубль, используйте RETURNING или проверяйте RowsAffected.
• Хранение message_id в отдельной таблице без уникального индекса - тогда дедупликация не работает.