隔离级别

概念定义

事务隔离级别规定一个事务可容忍哪些并发异常。经典 SQL 级别——读未提交、读已提交、可重复读、可串行化——依次禁止脏读、不可重复读、幻读，最终禁止一切串行化异常。更强的隔离消除更多异常，却会降低并发并增加中止。级别可按事务或会话选择，且理解你的数据库实际实现了什么很重要，因为各引擎并不相同。

为什么重要

这些级别所禁止的异常，恰恰是最难复现的缺陷：两个用户之间丢失的一次更新、报表中途变化的行数、被幻读插入破坏的约束。选择过弱的级别会让它们潜伏；选择过强的级别，例如对所有查询都用可串行化，则招致争用与中止，损害吞吐。级别是一种有意的取舍，而非可忽视的设定。

工程实践

识别事务必须保持的不变量，选择能保证它的最弱级别或显式锁——往往读已提交加乐观版本控制就足够。要了解你的引擎实际实现了什么，因为可重复读等知名名称在 MVCC 下表现不一。把更强级别带来的中止当作可重试的工作而非失败来处理，测试该级别允许其相互干扰的并发路径，并保持事务短小，使所选级别影响的空间更小。

代码示例

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

interface VersionedRow {
  readonly key: string;
  readonly value: string;
  readonly startTs: number;
  readonly endTs: number;
}

export class MvccStore {
  private versions: VersionedRow[] = [];
  private clock = 0;

  beginTransaction(): number {
    return ++this.clock;
  }

  write(key: string, value: string, ts: number): void {
    for (const row of this.versions) {
      if (row.key === key && row.endTs === Number.POSITIVE_INFINITY) row.endTs = ts;
    }
    this.versions.push({ key, value, startTs: ts, endTs: Number.POSITIVE_INFINITY });
  }

  read(key: string, snapshotTs: number): string | undefined {
    let found: VersionedRow | undefined;
    for (const row of this.versions) {
      if (row.key !== key) continue;
      if (row.startTs <= snapshotTs && snapshotTs < row.endTs) {
        if (!found || row.startTs > found.startTs) found = row;
      }
    }
    return found?.value;
  }
}