Зачем нужно выражение SELECT FOR UPDATE в SQL?

# SELECT FOR UPDATE в SQL

SELECT FOR UPDATE — это выражение, которое блокирует выбранные строки на время транзакции, предотвращая их изменение другими процессами. Это критично для обеспечения data consistency в конкурентной среде.

Основная проблема

Без блокировок может возникнуть race condition:

# Без FOR UPDATE — НЕПРАВИЛЬНО
connection = get_db_connection()

# Шаг 1: Читаем текущий баланс
user = connection.query("SELECT balance FROM users WHERE id = ?", (user_id,))
current_balance = user['balance']  # 1000

# Между шагом 1 и 2 другой процесс может изменить баланс!

# Шаг 2: Обновляем баланс
new_balance = current_balance - 100  # 900
connection.execute(
    "UPDATE users SET balance = ? WHERE id = ?",
    (new_balance, user_id)
)
connection.commit()

Сценарий race condition:

1. Процесс A читает balance = 1000
2. Процесс B читает balance = 1000
3. Процесс A снимает 100, записывает balance = 900
4. Процесс B снимает 100, записывает balance = 900
5. Итог: вместо 800 получилось 900 (потеря 100)

Решение: SELECT FOR UPDATE

# С FOR UPDATE — ПРАВИЛЬНО
from sqlalchemy import text
from sqlalchemy.orm import Session

def transfer_money(user_id: int, amount: float):
    session = Session(engine)
    
    try:
        # Читаем и блокируем запись
        user = session.query(User).with_for_update().filter(
            User.id == user_id
        ).one()
        
        # Теперь эта строка заблокирована
        # Другие процессы не могут изменять этого пользователя
        
        if user.balance >= amount:
            user.balance -= amount
            session.commit()
            return True
        else:
            session.rollback()
            return False
    
    finally:
        session.close()

Что происходит:

1. Процесс A читает и блокирует balance = 1000
2. Процесс B попытается прочитать и заблокировать — ждет
3. Процесс A обновляет, коммитит, открывает блокировку
4. Процесс B теперь читает balance = 900
5. Итог: 900 - 100 = 800 (правильно)

Варианты блокировок

1 SELECT FOR UPDATE (эксклюзивная блокировка)

BEGIN;
SELECT * FROM users WHERE id = 123 FOR UPDATE;
UPDATE users SET balance = balance - 100 WHERE id = 123;
COMMIT;

• Эксклюзивная блокировка (EXCLUSIVE LOCK)
• Никто не может читать ИЛИ писать эту строку
• Максимальная защита, но медленнее

2 SELECT FOR UPDATE SKIP LOCKED

BEGIN;
SELECT * FROM orders 
WHERE status = 'pending' AND worker_id IS NULL
FOR UPDATE SKIP LOCKED
LIMIT 5;

UPDATE orders 
SET worker_id = 42 
WHERE id IN (...);
COMMIT;

• Пропускает уже заблокированные строки
• Полезно для распределения работы между воркерами
• Не ждет, если строка заблокирована

3 SELECT FOR SHARE (read-only блокировка)

BEGIN;
SELECT * FROM users WHERE id = 123 FOR SHARE;
COMMIT;

• Несколько процессов могут читать одновременно
• Никто не может писать
• Слабее, чем FOR UPDATE

Практические примеры

Пример 1: Управление финансами

def withdraw_money(user_id: int, amount: float) -> bool:
    """Снять деньги со счета с гарантией консистентности."""
    from sqlalchemy import text
    
    with engine.begin() as connection:
        # Читаем и блокируем
        result = connection.execute(
            text("""
                SELECT balance FROM users 
                WHERE id = :user_id 
                FOR UPDATE
            """),
            {"user_id": user_id}
        )
        
        user = result.first()
        if not user or user[0] < amount:
            return False
        
        # Обновляем внутри той же транзакции
        connection.execute(
            text("UPDATE users SET balance = balance - :amount WHERE id = :user_id"),
            {"amount": amount, "user_id": user_id}
        )
        
        return True

Пример 2: Обработка очереди задач

def get_next_task() -> Task:
    """Получить следующую задачу для обработки."""
    from sqlalchemy import text
    
    with engine.begin() as connection:
        # Находим и блокируем свободную задачу
        result = connection.execute(
            text("""
                SELECT id, data FROM tasks 
                WHERE status = 'pending' 
                ORDER BY created_at 
                LIMIT 1 
                FOR UPDATE SKIP LOCKED
            """)
        )
        
        task = result.first()
        if not task:
            return None
        
        # Помечаем как в обработке
        connection.execute(
            text("UPDATE tasks SET status = 'processing' WHERE id = :task_id"),
            {"task_id": task[0]}
        )
        
        return Task(id=task[0], data=task[1])

Пример 3: Инвентаризация

def purchase_item(user_id: int, item_id: int, quantity: int) -> bool:
    """Купить товар, проверив наличие с блокировкой."""
    from sqlalchemy import text
    
    with engine.begin() as connection:
        # Читаем и блокируем товар
        result = connection.execute(
            text("""
                SELECT stock FROM inventory 
                WHERE item_id = :item_id 
                FOR UPDATE
            """),
            {"item_id": item_id}
        )
        
        item = result.first()
        if not item or item[0] < quantity:
            return False
        
        # Уменьшаем stock
        connection.execute(
            text("""
                UPDATE inventory 
                SET stock = stock - :qty 
                WHERE item_id = :item_id
            """),
            {"qty": quantity, "item_id": item_id}
        )
        
        # Создаем заказ
        connection.execute(
            text("""
                INSERT INTO orders (user_id, item_id, quantity) 
                VALUES (:user_id, :item_id, :qty)
            """),
            {"user_id": user_id, "item_id": item_id, "qty": quantity}
        )
        
        return True

Производительность и deadlocks

Потенциальная проблема: deadlock

# Процесс A
BEGIN;
SELECT * FROM users WHERE id = 1 FOR UPDATE;  # Блокирует user 1
SELECT * FROM users WHERE id = 2 FOR UPDATE;  # Ждет...

# Процесс B (в параллели)
BEGIN;
SELECT * FROM users WHERE id = 2 FOR UPDATE;  # Блокирует user 2
SELECT * FROM users WHERE id = 1 FOR UPDATE;  # Ждет...
# DEADLOCK!

Решение: всегда блокируй в одном порядке:

# Правильно
BEGIN;
SELECT * FROM users WHERE id IN (1, 2) ORDER BY id FOR UPDATE;

Когда использовать

✅ Используй FOR UPDATE:

• Финансовые операции
• Управление инвентарем
• Бронирование ресурсов
• Обработка конкурирующих запросов
• Критичные бизнес-операции

❌ Избегай FOR UPDATE:

• Простые SELECT запросы без изменений
• Высоконагруженные системы (может быть узким местом)
• Когда можно использовать optimistic locking

Альтернативы

Optimistic Locking (версионирование)

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    balance = Column(Float)
    version = Column(Integer, default=0)

def withdraw_money(user_id: int, amount: float):
    user = session.query(User).filter(User.id == user_id).one()
    
    if user.balance >= amount:
        user.balance -= amount
        user.version += 1  # Увеличиваем версию
        
        try:
            session.commit()
        except StaleDataError:
            # Кто-то еще обновил запись
            session.rollback()
            return False
    
    return True

Итог

SELECT FOR UPDATE — это мощный инструмент для обеспечения консистентности данных в конкурентной среде. Это особенно важно в микросервисной архитектуре, где множество процессов обращаются к одним данным. Правильное использование FOR UPDATE может спасти от критичных ошибок в логике бизнеса.