丢失更新

工作原理

事务 A 读取 balance = 100，事务 B 也读取 100。A 写入 110 并提交，B 写入 120 并提交。两笔 +10 入账后的正确余额是 120，结果恰好是 120，但那是因为两笔金额相同。若分别是 +10 和 +5，A 写入 110，B 写入 105，最终 105 完全丢失了 A 的入账。

异常的关键是读写序列，而不是行锁持有时长。如果读取发生在写锁之外，两个事务都可能基于同一个旧值写入。数据库的检测方式不同：PostgreSQL 可重复读会在第二次更新时抛出序列化错误，而某些引擎在某些隔离级别下允许覆盖。

预防

使用原子语句：UPDATE accounts SET balance = balance + 10 WHERE id = 7。数据库会串行化增量。或者使用乐观并发：增加 version 列，执行 WHERE id = 7 AND version = 3 的更新，若影响零行则重试。悲观锁 SELECT ... FOR UPDATE 可行，但会在应用逻辑期间持锁并可能死锁。当多行构成一个不变量时，可串行化隔离加应用重试是正确的。

何时选择

简单算术和计数器使用原子 SQL。人工编辑记录的表单使用乐观并发，让冲突显示给用户而不是静默重试。高争用的短临界区使用悲观锁。跨服务边界时，除非操作幂等且带版本，否则避免读改写。

权衡与示例

原子更新简单快速，但无法表达任意应用逻辑。乐观并发能检测冲突，却需要重试交互，在高争用下可能反复失败。悲观锁防止冲突，但降低吞吐并带来死锁风险。

示例：两名客服同时打开同一工单，各自通过替换整个标签列表来添加标签。第二次保存覆盖了第一名客服的标签。把标签存入独立表并加唯一约束，或给工单加版本号，可以保留两次编辑。整份保存 JSON 文档也有同样的问题。

代码示例

from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Any


class TransactionAborted(RuntimeError):
    pass


@dataclass
class Store:
    data: dict[str, Any] = field(default_factory=dict)
    commits: int = 0
    rollbacks: int = 0

    @contextmanager
    def transaction(self) -> Iterator["Transaction"]:
        snapshot = dict(self.data)
        tx = Transaction(self)
        try:
            yield tx
        except BaseException:
            self.data = snapshot
            self.rollbacks += 1
            raise
        else:
            self.commits += 1


@dataclass
class Transaction:
    store: Store

    def set(self, key: str, value: Any) -> None:
        self.store.data[key] = value

    def delete(self, key: str) -> None:
        self.store.data.pop(key, None)


def main() -> None:
    store = Store(data={"balance": 100})
    with store.transaction() as tx:
        tx.set("balance", 80)
        tx.set("currency", "USD")
    print(store.data, store.commits, store.rollbacks)

    try:
        with store.transaction() as tx:
            tx.set("balance", 0)
            raise TransactionAborted("payment declined")
    except TransactionAborted as error:
        print("rolled back:", error)
    print(store.data, store.commits, store.rollbacks)


if __name__ == "__main__":
    main()

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

engine = create_engine(
    'postgresql+psycopg://user:pass@localhost/reporting',
    isolation_level='READ COMMITTED',
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)

reporting_engine = engine.execution_options(
    isolation_level='REPEATABLE READ'
)


def transfer_funds(session: Session, src: int, dst: int, amount: int) -> None:
    session.execute(
        text('UPDATE accounts SET balance = balance - :amt WHERE id = :id'),
        {'amt': amount, 'id': src},
    )
    row = session.execute(
        text('SELECT balance FROM accounts WHERE id = :id'),
        {'id': src},
    ).scalar_one()
    if row < 0:
        raise ValueError('overdraft rejected')
    session.execute(
        text('UPDATE accounts SET balance = balance + :amt WHERE id = :id'),
        {'amt': amount, 'id': dst},
    )


with Session(engine) as session:
    with session.begin():
        transfer_funds(session, src=1, dst=2, amount=75)

with Session(reporting_engine) as session:
    snapshot = session.execute(
        text('SELECT sum(balance) FROM accounts')
    ).scalar_one()
    print('consistent snapshot total:', snapshot)