Senior

Hoe zag de route van optimistische blokkering eruit van het ontvangen van het bericht uit Kafka tot de update in de database?

sobes.tech AI

Antwoord van AI

Оптимистичная блокировка при обработке сообщений из Kafka и обновлении базы данных обычно выглядит так:

Получаем сообщение из Kafka, содержащее данные для обновления.
Извлекаем из сообщения ключ сущности и версию (или timestamp) текущей записи.
Выполняем запрос к базе данных с условием, что версия записи совпадает с той, что была в сообщении (т.е. никто не изменил запись с момента получения сообщения).
Если условие выполняется, обновляем запись, увеличивая версию.
Если условие не выполняется (конфликт версий), откатываем операцию или повторяем попытку.

Пример на Go с использованием SQL:

// msg содержит ID и версию записи
func processMessage(msg Message) error {
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()

    res, err := tx.Exec(`UPDATE items SET data = ?, version = version + 1 WHERE id = ? AND version = ?`, msg.Data, msg.ID, msg.Version)
    if err != nil {
        return err
    }

    rowsAffected, err := res.RowsAffected()
    if err != nil {
        return err
    }

    if rowsAffected == 0 {
        // Конфликт версий — кто-то обновил запись раньше
        return fmt.Errorf("optimistic lock failed")
    }

    return tx.Commit()
}

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

Vorige Volgende