多版本并发控制（MVCC）

概念定义

多版本并发控制为每个行版本按事务标识划定可见边界。读方重建事务开始时的快照、忽略更新版本；写方创建新版本并标记旧版死亡。纯读之间的锁彻底消失，不同隔离级别只在隐藏并发写的严格程度上不同。

工作机制

PostgreSQL 元组携带 xmin 与 xmax 事务号；InnoDB 把变更写入回滚段并使用读视图。快照记录在飞事务集合，行可见性逐行计算。隔离级别改变快照策略：每语句一快照，或每事务一快照。

使用场景

OLTP 与报表同服、高并发读、任何读锁会串行化吞吐的场景。应把 vacuum 与 undo 保留当作一等运维对象来规划，而不是依赖默认值。

常见误区

MVCC 只消除读写阻塞，同行并发写依旧冲突；它也不是默认的可串行化。长事务持有的快照会钉住垃圾回收，在 PostgreSQL 造成表膨胀与事务号回卷风险，在 InnoDB 堆积巨型 undo 日志。快照年龄与死元组数应作为容量信号监控。

代码示例

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)

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()