事务管理与并发控制

目录

• 事务管理与并发控制：构建可靠数据系统的核心技术
• • 1. 引言
  • • 1.1 什么是事务？
    • 1.2 为什么需要并发控制？
  • 2. ACID属性：事务的基石
  • • 2.1 原子性（Atomicity）
    • 2.2 一致性（Consistency）
    • 2.3 隔离性（Isolation）
    • 2.4 持久性（Durability）
  • 3. 并发控制问题
  • • 3.1 脏读（Dirty Read）
    • 3.2 不可重复读（Non-repeatable Read）
    • 3.3 幻读（Phantom Read）
    • 3.4 丢失更新（Lost Update）
  • 4. 事务隔离级别
  • • 4.1 SQL标准隔离级别
    • 4.2 隔离级别的实现
    • • 4.2.1 读已提交的实现
      • 4.2.2 可重复读的实现
      • 4.2.3 串行化的实现
  • 5. 并发控制技术
  • • 5.1 锁机制（悲观并发控制）
    • • 5.1.1 锁的类型
      • 5.1.2 锁的兼容性矩阵
      • 5.1.3 两阶段锁协议（2PL）
    • 5.2 时间戳排序（Timestamp Ordering）
    • • 5.2.1 基本规则
    • 5.3 多版本并发控制（MVCC）
    • • 5.3.1 MVCC工作原理
      • 5.3.2 版本选择规则
    • 5.4 乐观并发控制（OCC）
    • • 5.4.1 三个阶段
      • 5.4.2 验证方法
  • 6. 死锁处理
  • • 6.1 死锁检测
    • • 6.1.1 等待图（Wait-for Graph）
      • 6.1.2 超时机制
    • 6.2 死锁预防
    • • 6.2.1 资源排序法
      • 6.2.2 时间戳死锁预防
    • 6.3 死锁避免
    • • 6.3.1 银行家算法（Banker's Algorithm）
  • 7. Python实现：事务管理与并发控制框架
  • 8. 分布式事务
  • • 8.1 两阶段提交（2PC）
    • • 8.1.1 阶段一：准备阶段
      • 8.1.2 阶段二：提交阶段
    • 8.2 三阶段提交（3PC）
    • 8.3 补偿事务（Saga模式）
  • 9. 现代并发控制技术
  • • 9.1 无锁数据结构
    • 9.2 软件事务内存（STM）
  • 10. 性能优化与最佳实践
  • • 10.1 锁粒度优化
    • • 10.1.1 锁粒度选择
      • 10.1.2 锁升级策略
    • 10.2 事务设计最佳实践
    • 10.3 分布式事务最佳实践
  • 11. 监控与诊断
  • • 11.1 关键性能指标
    • 11.2 诊断工具
  • 12. 未来趋势与挑战
  • • 12.1 新硬件架构的影响
    • 12.2 新型并发控制算法
    • 12.3 云原生事务处理
  • 13. 总结
  • • 13.1 关键要点
    • 13.2 实践建议

『宝藏代码胶囊开张啦！』—— 我的 CodeCapsule 来咯！✨写代码不再头疼！我的新站点 CodeCapsule 主打一个 “白菜价”+“量身定制”！无论是卡脖子的毕设/课设/文献复现，需要灵光一现的算法改进，还是想给项目加个“外挂”，这里都有便宜又好用的代码方案等你发现！低成本，高适配，助你轻松通关！速来围观 👉 CodeCapsule官网

事务管理与并发控制：构建可靠数据系统的核心技术

1. 引言

在现代多用户数据库系统中，事务管理和并发控制是确保数据一致性、完整性和系统可靠性的核心技术。随着微服务架构和分布式系统的普及，这些技术的重要性愈发凸显。一个设计良好的事务系统能够确保即使在系统故障或并发访问的情况下，数据依然保持一致性。

1.1 什么是事务？

事务是数据库管理系统中的一个逻辑工作单元，它包含一系列操作，这些操作要么全部成功执行，要么全部不执行。事务必须满足ACID属性，这是事务处理系统的基石。

#mermaid-svg-2Kiy7d4Tb8WsXzWj{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .error-icon{fill:#552222;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .marker{fill:#333333;stroke:#333333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .marker.cross{stroke:#333333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-2Kiy7d4Tb8WsXzWj p{margin:0;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .cluster-label text{fill:#333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .cluster-label span{color:#333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .cluster-label span p{background-color:transparent;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .label text,#mermaid-svg-2Kiy7d4Tb8WsXzWj span{fill:#333;color:#333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .node rect,#mermaid-svg-2Kiy7d4Tb8WsXzWj .node circle,#mermaid-svg-2Kiy7d4Tb8WsXzWj .node ellipse,#mermaid-svg-2Kiy7d4Tb8WsXzWj .node polygon,#mermaid-svg-2Kiy7d4Tb8WsXzWj .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .rough-node .label text,#mermaid-svg-2Kiy7d4Tb8WsXzWj .node .label text,#mermaid-svg-2Kiy7d4Tb8WsXzWj .image-shape .label,#mermaid-svg-2Kiy7d4Tb8WsXzWj .icon-shape .label{text-anchor:middle;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .rough-node .label,#mermaid-svg-2Kiy7d4Tb8WsXzWj .node .label,#mermaid-svg-2Kiy7d4Tb8WsXzWj .image-shape .label,#mermaid-svg-2Kiy7d4Tb8WsXzWj .icon-shape .label{text-align:center;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .node.clickable{cursor:pointer;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .arrowheadPath{fill:#333333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-2Kiy7d4Tb8WsXzWj .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-2Kiy7d4Tb8WsXzWj .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-2Kiy7d4Tb8WsXzWj .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .cluster text{fill:#333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .cluster span{color:#333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-2Kiy7d4Tb8WsXzWj rect.text{fill:none;stroke-width:0;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .icon-shape,#mermaid-svg-2Kiy7d4Tb8WsXzWj .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .icon-shape p,#mermaid-svg-2Kiy7d4Tb8WsXzWj .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .icon-shape rect,#mermaid-svg-2Kiy7d4Tb8WsXzWj .image-shape rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-2Kiy7d4Tb8WsXzWj .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-2Kiy7d4Tb8WsXzWj .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-2Kiy7d4Tb8WsXzWj :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

是

否

事务开始

操作1

操作2

操作N

所有操作成功?

提交事务

回滚事务

数据持久化

数据恢复原状

1.2 为什么需要并发控制？

在多用户环境中，多个事务可能同时访问相同的数据资源。没有适当的并发控制，可能会导致以下问题：

2. ACID属性：事务的基石

2.1 原子性（Atomicity）

原子性确保事务中的所有操作要么全部完成，要么全部不执行。这是通过事务日志和回滚机制实现的。

数学表示：设事务

T

T

T包含操作集合

O

1

,

O

2

,

.

.

.

,

O

n

{O_1, O_2, …, O_n}

O1​,O2​,…,On​，则原子性要求：

要么 

∀

O

i

∈

T

,

O

i

 成功

\\text{要么 } \\forall O_i \\in T, O_i \\text{ 成功}

要么 ∀Oi​∈T,Oi​ 成功

要么 

∀

O

i

∈

T

,

O

i

 回滚

\\text{要么 } \\forall O_i \\in T, O_i \\text{ 回滚}

要么 ∀Oi​∈T,Oi​ 回滚

2.2 一致性（Consistency）

一致性确保事务将数据库从一个一致状态转换到另一个一致状态。这包括所有完整性约束、业务规则和逻辑约束。

数学表示：设

S

b

S_b

Sb​是事务开始前的数据库状态，

S

a

S_a

Sa​是事务结束后的数据库状态，

C

C

C是一致性约束集合，则：

∀

c

∈

C

,

c

(

S

b

)

=

true

⇒

c

(

S

a

)

=

true

\\forall c \\in C, c(S_b) = \\text{true} \\Rightarrow c(S_a) = \\text{true}

∀c∈C,c(Sb​)=true⇒c(Sa​)=true

2.3 隔离性（Isolation）

隔离性确保并发执行的事务相互隔离，每个事务都感觉不到其他事务的存在。

2.4 持久性（Durability）

持久性确保一旦事务提交，其结果将永久保存在数据库中，即使系统发生故障。

3. 并发控制问题

3.1 脏读（Dirty Read）

脏读发生在一个事务读取了另一个未提交事务修改的数据。如果该未提交事务随后回滚，读取到的数据就是无效的。

# 脏读示例
"""
事务A: UPDATE accounts SET balance = balance – 100 WHERE id = 1
事务B: SELECT balance FROM accounts WHERE id = 1 # 读取未提交数据
事务A: ROLLBACK # 事务A回滚
# 事务B读取到了不存在的数据
"""

3.2 不可重复读（Non-repeatable Read）

不可重复读发生在一个事务内两次读取同一数据，但得到的结果不同，因为另一个事务在这期间修改了该数据。

# 不可重复读示例
"""
事务A: SELECT balance FROM accounts WHERE id = 1 # 返回1000
事务B: UPDATE accounts SET balance = 900 WHERE id = 1
事务A: SELECT balance FROM accounts WHERE id = 1 # 返回900，与第一次读取不同
"""

3.3 幻读（Phantom Read）

幻读发生在一个事务两次执行相同的查询，但返回的行数不同，因为另一个事务在这期间插入了新行。

# 幻读示例
"""
事务A: SELECT COUNT(*) FROM accounts WHERE balance > 1000 # 返回5
事务B: INSERT INTO accounts (balance) VALUES (2000)
事务A: SELECT COUNT(*) FROM accounts WHERE balance > 1000 # 返回6，出现了"幻影行"
"""

3.4 丢失更新（Lost Update）

丢失更新发生在两个事务同时读取并修改同一数据，后提交的事务覆盖了先提交事务的修改。

# 丢失更新示例
"""
事务A: 读取balance=1000，计算balance=900，更新balance=900
事务B: 读取balance=1000，计算balance=1100，更新balance=1100
# 事务B的更新覆盖了事务A的更新
"""

4. 事务隔离级别

4.1 SQL标准隔离级别

SQL标准定义了四个隔离级别，从低到高分别是：

#mermaid-svg-WcCFGT6HwFyGpqrZ{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-WcCFGT6HwFyGpqrZ .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-WcCFGT6HwFyGpqrZ .error-icon{fill:#552222;}#mermaid-svg-WcCFGT6HwFyGpqrZ .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-WcCFGT6HwFyGpqrZ .marker{fill:#333333;stroke:#333333;}#mermaid-svg-WcCFGT6HwFyGpqrZ .marker.cross{stroke:#333333;}#mermaid-svg-WcCFGT6HwFyGpqrZ svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-WcCFGT6HwFyGpqrZ p{margin:0;}#mermaid-svg-WcCFGT6HwFyGpqrZ .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-WcCFGT6HwFyGpqrZ .cluster-label text{fill:#333;}#mermaid-svg-WcCFGT6HwFyGpqrZ .cluster-label span{color:#333;}#mermaid-svg-WcCFGT6HwFyGpqrZ .cluster-label span p{background-color:transparent;}#mermaid-svg-WcCFGT6HwFyGpqrZ .label text,#mermaid-svg-WcCFGT6HwFyGpqrZ span{fill:#333;color:#333;}#mermaid-svg-WcCFGT6HwFyGpqrZ .node rect,#mermaid-svg-WcCFGT6HwFyGpqrZ .node circle,#mermaid-svg-WcCFGT6HwFyGpqrZ .node ellipse,#mermaid-svg-WcCFGT6HwFyGpqrZ .node polygon,#mermaid-svg-WcCFGT6HwFyGpqrZ .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-WcCFGT6HwFyGpqrZ .rough-node .label text,#mermaid-svg-WcCFGT6HwFyGpqrZ .node .label text,#mermaid-svg-WcCFGT6HwFyGpqrZ .image-shape .label,#mermaid-svg-WcCFGT6HwFyGpqrZ .icon-shape .label{text-anchor:middle;}#mermaid-svg-WcCFGT6HwFyGpqrZ .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-WcCFGT6HwFyGpqrZ .rough-node .label,#mermaid-svg-WcCFGT6HwFyGpqrZ .node .label,#mermaid-svg-WcCFGT6HwFyGpqrZ .image-shape .label,#mermaid-svg-WcCFGT6HwFyGpqrZ .icon-shape .label{text-align:center;}#mermaid-svg-WcCFGT6HwFyGpqrZ .node.clickable{cursor:pointer;}#mermaid-svg-WcCFGT6HwFyGpqrZ .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-WcCFGT6HwFyGpqrZ .arrowheadPath{fill:#333333;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-WcCFGT6HwFyGpqrZ .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-WcCFGT6HwFyGpqrZ .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-WcCFGT6HwFyGpqrZ .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-WcCFGT6HwFyGpqrZ .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-WcCFGT6HwFyGpqrZ .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-WcCFGT6HwFyGpqrZ .cluster text{fill:#333;}#mermaid-svg-WcCFGT6HwFyGpqrZ .cluster span{color:#333;}#mermaid-svg-WcCFGT6HwFyGpqrZ div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-WcCFGT6HwFyGpqrZ .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-WcCFGT6HwFyGpqrZ rect.text{fill:none;stroke-width:0;}#mermaid-svg-WcCFGT6HwFyGpqrZ .icon-shape,#mermaid-svg-WcCFGT6HwFyGpqrZ .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-WcCFGT6HwFyGpqrZ .icon-shape p,#mermaid-svg-WcCFGT6HwFyGpqrZ .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-WcCFGT6HwFyGpqrZ .icon-shape rect,#mermaid-svg-WcCFGT6HwFyGpqrZ .image-shape rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-WcCFGT6HwFyGpqrZ .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-WcCFGT6HwFyGpqrZ .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-WcCFGT6HwFyGpqrZ :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

防止脏读

防止不可重复读

防止幻读

读未提交

读已提交

可重复读

串行化

4.2 隔离级别的实现

4.2.1 读已提交的实现

大多数数据库默认使用读已提交隔离级别，通过以下机制实现：

数学表示：设

T

i

T_i

Ti​是事务，

x

x

x是数据项，

T

S

(

T

i

)

TS(T_i)

TS(Ti​)是事务时间戳

• 读操作：读取$TS(x)\le TS(T_i)$
• 写操作：创建新版本，时间戳为$TS(T_i)$

4.2.2 可重复读的实现

可重复读通常通过以下方式实现：

4.2.3 串行化的实现

串行化通过以下机制实现：

5. 并发控制技术

5.1 锁机制（悲观并发控制）

5.1.1 锁的类型

5.1.2 锁的兼容性矩阵

5.1.3 两阶段锁协议（2PL）

两阶段锁协议确保事务的可串行化，包含两个阶段：

数学表示： 设

T

i

T_i

Ti​是事务，

l

o

c

k

i

(

x

)

lock_i(x)

locki​(x)表示

T

i

T_i

Ti​在

x

x

x上加锁，

u

n

l

o

c

k

i

(

x

)

unlock_i(x)

unlocki​(x)表示释放锁 则对于任意事务

T

i

T_i

Ti​，存在一个时间点

c

i

c_i

ci​（提交点），使得：

• 对于所有$loc{k}_i(x)$
• 对于所有$unloc{k}_i(x)$

5.2 时间戳排序（Timestamp Ordering）

时间戳排序协议为每个事务分配唯一的时间戳，确保事务按时间戳顺序执行。

5.2.1 基本规则

对于数据项

x

x

x，维护：

• $R\text{–}ts(x)$
• $W\text{–}ts(x)$

规则：

5.3 多版本并发控制（MVCC）

MVCC通过为数据项维护多个版本来提高并发性能。

5.3.1 MVCC工作原理

每个数据项

x

x

x有多个版本：

x

1

,

x

2

,

.

.

.

,

x

n

x_1, x_2, …, x_n

x1​,x2​,…,xn​ 每个版本包含：

• 数据值
• 创建时间戳（start-TS）
• 删除/过期时间戳（end-TS）

5.3.2 版本选择规则

事务

T

i

T_i

Ti​（时间戳为

T

S

(

T

i

)

TS(T_i)

TS(Ti​)）读取

x

x

x时，选择满足以下条件的版本

x

k

x_k

xk​：

start-TS

(

x

k

)

≤

T

S

(

T

i

)

<

end-TS

(

x

k

)

\\text{start-TS}(x_k) \\leq TS(T_i) < \\text{end-TS}(x_k)

start-TS(xk​)≤TS(Ti​)<end-TS(xk​)

5.4 乐观并发控制（OCC）

乐观并发控制假设冲突很少发生，在事务提交时才检查冲突。

5.4.1 三个阶段

5.4.2 验证方法

数学表示： 设

R

S

(

T

i

)

RS(T_i)

RS(Ti​)是事务

T

i

T_i

Ti​的读集，

W

S

(

T

i

)

WS(T_i)

WS(Ti​)是写集 事务

T

i

T_i

Ti​和

T

j

T_j

Tj​冲突的条件是：

R

S

(

T

i

)

∩

W

S

(

T

j

)

≠

∅

或

W

S

(

T

i

)

∩

R

S

(

T

j

)

≠

∅

RS(T_i) \\cap WS(T_j) \\neq \\emptyset \\quad \\text{或} \\quad WS(T_i) \\cap RS(T_j) \\neq \\emptyset

RS(Ti​)∩WS(Tj​)=∅或WS(Ti​)∩RS(Tj​)=∅

6. 死锁处理

6.1 死锁检测

6.1.1 等待图（Wait-for Graph）

等待图

G

=

(

V

,

E

)

G = (V, E)

G=(V,E)，其中：

• $V$
• $E$

死锁检测算法定期检查等待图中是否存在环。

# 等待图检测死锁的简化实现
def detect_deadlock(wait_for_graph):
"""检测等待图中是否存在环（死锁）"""
def dfs(node, visited, recursion_stack):
visited[node] = True
recursion_stack[node] = True

for neighbor in wait_for_graph.get(node, []):
if not visited[neighbor]:
if dfs(neighbor, visited, recursion_stack):
return True
elif recursion_stack[neighbor]:
return True

recursion_stack[node] = False
return False

nodes = list(wait_for_graph.keys())
visited = {node: False for node in nodes}
recursion_stack = {node: False for node in nodes}

for node in nodes:
if not visited[node]:
if dfs(node, visited, recursion_stack):
return True
return False

6.1.2 超时机制

为事务设置超时时间，超过时间则假定可能发生死锁，进行回滚。

6.2 死锁预防

6.2.1 资源排序法

对所有资源进行全序排序，要求事务按顺序申请资源。

6.2.2 时间戳死锁预防

6.3 死锁避免

6.3.1 银行家算法（Banker’s Algorithm）

将系统视为银行家，资源视为资金，只有确保不会导致死锁时才分配资源。

数学表示： 设系统有

m

m

m类资源，每类资源有

R

j

R_j

Rj​个实例

• $Available[j]$
• $Max[i][j]$
• $Allocation[i][j]$
• $Need[i][j]=Max[i][j]-Allocation[i][j]$

安全性算法检查是否存在安全序列。

7. Python实现：事务管理与并发控制框架

以下是完整的事务管理与并发控制框架实现：

"""
事务管理与并发控制框架
提供完整的事务管理、锁管理、死锁检测和并发控制功能
"""

import threading
import time
import uuid
from datetime import datetime
from enum import Enum, auto
from typing import Dict, List, Set, Optional, Any, Tuple
from dataclasses import dataclass, field
from collections import defaultdict, deque
import logging
from contextlib import contextmanager
import random

# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
)
logger = logging.getLogger(__name__)

class IsolationLevel(Enum):
"""事务隔离级别"""
READ_UNCOMMITTED = "READ_UNCOMMITTED"
READ_COMMITTED = "READ_COMMITTED"
REPEATABLE_READ = "REPEATABLE_READ"
SERIALIZABLE = "SERIALIZABLE"

class LockType(Enum):
"""锁类型"""
SHARED = "SHARED" # 共享锁
EXCLUSIVE = "EXCLUSIVE" # 排他锁
INTENTION_SHARED = "INTENTION_SHARED"
INTENTION_EXCLUSIVE = "INTENTION_EXCLUSIVE"
SHARED_INTENTION_EXCLUSIVE = "SHARED_INTENTION_EXCLUSIVE"

class TransactionStatus(Enum):
"""事务状态"""
ACTIVE = "ACTIVE"
COMMITTED = "COMMITTED"
ABORTED = "ABORTED"
BLOCKED = "BLOCKED"

@dataclass
class LockRequest:
"""锁请求"""
transaction_id: str
lock_type: LockType
timestamp: float
granted: bool = False

@dataclass
class DataItem:
"""数据项"""
key: str
value: Any
version: int = 1
created_at: float = field(default_factory=time.time)
updated_at: float = field(default_factory=time.time)

def update(self, new_value: Any):
"""更新数据项"""
self.value = new_value
self.version += 1
self.updated_at = time.time()

@dataclass
class Transaction:
"""事务"""
transaction_id: str
timestamp: float
isolation_level: IsolationLevel
status: TransactionStatus = TransactionStatus.ACTIVE
read_set: Set[str] = field(default_factory=set) # 读取的数据项集合
write_set: Set[str] = field(default_factory=set) # 写入的数据项集合
lock_requests: Dict[str, LockType] = field(default_factory=dict) # 锁请求
snapshot: Dict[str, DataItem] = field(default_factory=dict) # 快照（用于可重复读）

def __repr__(self):
return (f"Transaction(id={self.transaction_id[:8]}, "
f"ts={self.timestamp:.2f}, status={self.status.name})")

class LockManager:
"""
锁管理器
管理数据项的锁分配和释放
"""

# 锁兼容性矩阵
LOCK_COMPATIBILITY = {
LockType.SHARED: {
LockType.SHARED: True,
LockType.EXCLUSIVE: False,
LockType.INTENTION_SHARED: True,
LockType.INTENTION_EXCLUSIVE: False,
LockType.SHARED_INTENTION_EXCLUSIVE: False,
},
LockType.EXCLUSIVE: {
LockType.SHARED: False,
LockType.EXCLUSIVE: False,
LockType.INTENTION_SHARED: False,
LockType.INTENTION_EXCLUSIVE: False,
LockType.SHARED_INTENTION_EXCLUSIVE: False,
},
LockType.INTENTION_SHARED: {
LockType.SHARED: True,
LockType.EXCLUSIVE: False,
LockType.INTENTION_SHARED: True,
LockType.INTENTION_EXCLUSIVE: True,
LockType.SHARED_INTENTION_EXCLUSIVE: True,
},
LockType.INTENTION_EXCLUSIVE: {
LockType.SHARED: False,
LockType.EXCLUSIVE: False,
LockType.INTENTION_SHARED: True,
LockType.INTENTION_EXCLUSIVE: True,
LockType.SHARED_INTENTION_EXCLUSIVE: False,
},
LockType.SHARED_INTENTION_EXCLUSIVE: {
LockType.SHARED: False,
LockType.EXCLUSIVE: False,
LockType.INTENTION_SHARED: True,
LockType.INTENTION_EXCLUSIVE: False,
LockType.SHARED_INTENTION_EXCLUSIVE: False,
},
}

def __init__(self):
"""初始化锁管理器"""
self.locks: Dict[str, List[LockRequest]] = defaultdict(list)
self.lock_holders: Dict[str, Dict[str, LockType]] = defaultdict(dict)
self.wait_for_graph: Dict[str, Set[str]] = defaultdict(set)
self.lock = threading.RLock() # 用于线程安全

def request_lock(
self,
transaction_id: str,
data_item: str,
lock_type: LockType,
timeout: float = 5.0
) –> bool:
"""
请求锁

Args:
transaction_id: 事务ID
data_item: 数据项
lock_type: 锁类型
timeout: 超时时间（秒）

Returns:
是否成功获取锁
"""
with self.lock:
# 检查是否已经持有相同或更强的锁
current_lock = self.lock_holders[data_item].get(transaction_id)
if current_lock and self._is_lock_upgrade(current_lock, lock_type):
# 锁升级
return self._upgrade_lock(transaction_id, data_item, current_lock, lock_type, timeout)

# 创建锁请求
lock_request = LockRequest(
transaction_id=transaction_id,
lock_type=lock_type,
timestamp=time.time()
)

# 添加到等待队列
self.locks[data_item].append(lock_request)

# 尝试获取锁
return self._try_grant_locks(data_item, timeout)

def _is_lock_upgrade(self, current_lock: LockType, requested_lock: LockType) –> bool:
"""检查是否是锁升级"""
# 锁强度顺序：S < IS < IX < SIX < X
lock_strength = {
LockType.SHARED: 1,
LockType.INTENTION_SHARED: 2,
LockType.INTENTION_EXCLUSIVE: 3,
LockType.SHARED_INTENTION_EXCLUSIVE: 4,
LockType.EXCLUSIVE: 5,
}
return lock_strength[requested_lock] > lock_strength[current_lock]

def _upgrade_lock(
self,
transaction_id: str,
data_item: str,
current_lock: LockType,
new_lock: LockType,
timeout: float
) –> bool:
"""升级锁"""
# 暂时释放当前锁
self.release_lock(transaction_id, data_item, current_lock)

# 请求新锁
success = self.request_lock(transaction_id, data_item, new_lock, timeout)

if not success:
# 升级失败，重新获取原锁
self.request_lock(transaction_id, data_item, current_lock, timeout)

return success

def _try_grant_locks(self, data_item: str, timeout: float) –> bool:
"""尝试授予锁"""
start_time = time.time()

while time.time() – start_time < timeout:
with self.lock:
# 检查每个等待的锁请求
for lock_request in self.locks[data_item]:
if not lock_request.granted:
if self._can_grant_lock(data_item, lock_request):
# 授予锁
lock_request.granted = True
self.lock_holders[data_item][lock_request.transaction_id] = lock_request.lock_type

# 更新等待图（移除等待关系）
for other_request in self.locks[data_item]:
if (not other_request.granted and
other_request.transaction_id != lock_request.transaction_id):
self.wait_for_graph[other_request.transaction_id].add(
lock_request.transaction_id
)

logger.info(f"授予锁: {lock_request.transaction_id[:8]} -> {data_item} "
f"({lock_request.lock_type.name})")
return True

# 等待一段时间再重试
time.sleep(0.01)

# 超时，移除锁请求
with self.lock:
self.locks[data_item] = [lr for lr in self.locks[data_item] if lr.granted]

logger.warning(f"锁请求超时: {data_item}")
return False

def _can_grant_lock(self, data_item: str, lock_request: LockRequest) –> bool:
"""检查是否可以授予锁"""
# 检查与已授予锁的兼容性
for holder_id, holder_lock in self.lock_holders[data_item].items():
if holder_id != lock_request.transaction_id:
if not self.LOCK_COMPATIBILITY[lock_request.lock_type][holder_lock]:
return False

# 检查与等待锁的兼容性（防止饿死）
for waiting_request in self.locks[data_item]:
if (waiting_request.transaction_id != lock_request.transaction_id and
waiting_request.timestamp < lock_request.timestamp and
not self.LOCK_COMPATIBILITY[lock_request.lock_type][waiting_request.lock_type]):
return False

return True

def release_lock(self, transaction_id: str, data_item: str, lock_type: Optional[LockType] = None):
"""
释放锁

Args:
transaction_id: 事务ID
data_item: 数据项
lock_type: 锁类型（如果为None，释放所有类型的锁）
"""
with self.lock:
if data_item in self.lock_holders:
if lock_type:
if (transaction_id in self.lock_holders[data_item] and
self.lock_holders[data_item][transaction_id] == lock_type):
del self.lock_holders[data_item][transaction_id]
logger.info(f"释放锁: {transaction_id[:8]} -> {data_item} ({lock_type.name})")
else:
if transaction_id in self.lock_holders[data_item]:
del self.lock_holders[data_item][transaction_id]
logger.info(f"释放所有锁: {transaction_id[:8]} -> {data_item}")

# 清理等待队列
if data_item in self.locks:
self.locks[data_item] = [
lr for lr in self.locks[data_item]
if lr.transaction_id != transaction_id
]

# 清理等待图
if transaction_id in self.wait_for_graph:
del self.wait_for_graph[transaction_id]

# 移除其他事务对该事务的等待
for waits in self.wait_for_graph.values():
if transaction_id in waits:
waits.remove(transaction_id)

def release_all_locks(self, transaction_id: str):
"""释放事务的所有锁"""
with self.lock:
for data_item in list(self.lock_holders.keys()):
if transaction_id in self.lock_holders[data_item]:
del self.lock_holders[data_item][transaction_id]

# 清理所有等待队列
for data_item in self.locks:
self.locks[data_item] = [
lr for lr in self.locks[data_item]
if lr.transaction_id != transaction_id
]

# 清理等待图
if transaction_id in self.wait_for_graph:
del self.wait_for_graph[transaction_id]

for waits in self.wait_for_graph.values():
if transaction_id in waits:
waits.remove(transaction_id)

def detect_deadlock(self) –> Optional[List[str]]:
"""
检测死锁

Returns:
死锁环中的事务ID列表，如果没有死锁返回None
"""
with self.lock:
# 使用深度优先搜索检测环
def dfs(node, visited, stack, path):
visited.add(node)
stack.add(node)
path.append(node)

for neighbor in self.wait_for_graph.get(node, set()):
if neighbor not in visited:
if dfs(neighbor, visited, stack, path):
return True
elif neighbor in stack:
# 找到环
cycle_start = path.index(neighbor)
return path[cycle_start:]

stack.remove(node)
path.pop()
return False

visited = set()
stack = set()

for node in list(self.wait_for_graph.keys()):
if node not in visited:
path = []
result = dfs(node, visited, stack, path)
if result:
return result if isinstance(result, list) else None

return None

def get_lock_status(self) –> Dict[str, Any]:
"""获取锁状态"""
with self.lock:
return {
"locks": {
item: [
{
"transaction": lr.transaction_id[:8],
"type": lr.lock_type.name,
"granted": lr.granted
}
for lr in requests
]
for item, requests in self.locks.items()
},
"holders": {
item: {
tid[:8]: lock_type.name
for tid, lock_type in holders.items()
}
for item, holders in self.lock_holders.items()
},
"wait_for_graph": {
tid[:8]: [w[:8] for w in waits]
for tid, waits in self.wait_for_graph.items()
}
}

class ConcurrencyControlManager:
"""
并发控制管理器
实现多种并发控制协议
"""

def __init__(self):
"""初始化并发控制管理器"""
self.lock_manager = LockManager()
self.timestamp_counter = 0
self.timestamp_lock = threading.Lock()
self.data_versions: Dict[str, List[DataItem]] = defaultdict(list) # MVCC版本链
self.last_committed: Dict[str, DataItem] = {} # 最后提交的版本

def get_next_timestamp(self) –> float:
"""获取下一个时间戳"""
with self.timestamp_lock:
self.timestamp_counter += 1
return float(self.timestamp_counter)

def lock_based_control(
self,
transaction: Transaction,
data_item: str,
operation: str, # "read" 或 "write"
timeout: float = 5.0
) –> bool:
"""
基于锁的并发控制

Args:
transaction: 事务
data_item: 数据项
operation: 操作类型
timeout: 超时时间

Returns:
是否成功
"""
# 根据操作类型确定需要的锁
if operation == "read":
lock_type = LockType.SHARED
else: # write
lock_type = LockType.EXCLUSIVE

# 请求锁
success = self.lock_manager.request_lock(
transaction.transaction_id,
data_item,
lock_type,
timeout
)

if success:
# 记录锁请求
transaction.lock_requests[data_item] = lock_type

if operation == "read":
transaction.read_set.add(data_item)
else:
transaction.write_set.add(data_item)

return success

def optimistic_control(
self,
transaction: Transaction,
data_item: str,
operation: str,
value: Any = None
) –> bool:
"""
乐观并发控制

Args:
transaction: 事务
data_item: 数据项
operation: 操作类型
value: 写入的值（仅用于写操作）

Returns:
是否成功
"""
if operation == "read":
# 读取到事务的私有工作区
if data_item in self.last_committed:
# 保存读取的数据和版本
item = self.last_committed[data_item]
transaction.snapshot[data_item] = DataItem(
key=item.key,
value=item.value,
version=item.version
)
transaction.read_set.add(data_item)
return True
return False

else: # write
# 写入到事务的私有工作区
if data_item in transaction.snapshot:
# 更新已有项
transaction.snapshot[data_item].value = value
else:
# 创建新项
transaction.snapshot[data_item] = DataItem(
key=data_item,
value=value,
version=1
)
transaction.write_set.add(data_item)
return True

def validate_optimistic_transaction(self, transaction: Transaction) –> bool:
"""
验证乐观事务

Args:
transaction: 事务

Returns:
是否通过验证
"""
# 向后验证：检查读取的数据是否被其他事务修改
for data_item in transaction.read_set:
if data_item in self.last_committed:
committed_version = self.last_committed[data_item].version
snapshot_version = transaction.snapshot[data_item].version

if committed_version > snapshot_version:
logger.warning(f"验证失败: {data_item} 已被修改")
return False

# 向前验证：检查写入的数据是否影响其他事务的读取
# 这里简化处理，实际需要检查其他活跃事务
return True

def commit_optimistic_transaction(self, transaction: Transaction) –> bool:
"""
提交乐观事务

Args:
transaction: 事务

Returns:
是否成功提交
"""
# 验证
if not self.validate_optimistic_transaction(transaction):
return False

# 写入数据
for data_item in transaction.write_set:
if data_item in transaction.snapshot:
item = transaction.snapshot[data_item]
item.version += 1
item.updated_at = time.time()
self.last_committed[data_item] = item

return True

def mvcc_read(
self,
transaction: Transaction,
data_item: str
) –> Optional[Any]:
"""
MVCC读取

Args:
transaction: 事务
data_item: 数据项

Returns:
读取的值或None
"""
if data_item not in self.data_versions:
return None

# 找到适合事务时间戳的版本
# 选择满足 start-TS <= transaction.timestamp < end-TS 的版本
versions = self.data_versions[data_item]

for version in reversed(versions): # 从最新版本开始查找
# 简化：假设版本时间戳就是创建时间
if version.created_at <= transaction.timestamp:
transaction.read_set.add(data_item)
return version.value

return None

def mvcc_write(
self,
transaction: Transaction,
data_item: str,
value: Any
) –> DataItem:
"""
MVCC写入

Args:
transaction: 事务
data_item: 数据项
value: 值

Returns:
创建的版本
"""
# 创建新版本
new_version = DataItem(
key=data_item,
value=value,
version=len(self.data_versions.get(data_item, [])) + 1,
created_at=transaction.timestamp
)

# 添加到版本链
self.data_versions[data_item].append(new_version)
transaction.write_set.add(data_item)

return new_version

def detect_and_resolve_deadlock(self) –> Optional[str]:
"""
检测并解决死锁

Returns:
被终止的事务ID，如果没有死锁返回None
"""
deadlock_cycle = self.lock_manager.detect_deadlock()

if deadlock_cycle:
logger.warning(f"检测到死锁: {' -> '.join(tid[:8] for tid in deadlock_cycle)}")

# 选择牺牲者（这里选择时间戳最大的事务，即最年轻的事务）
victim = max(deadlock_cycle, key=lambda tid: float(tid.split('_')[1]))

logger.info(f"选择事务 {victim[:8]} 作为牺牲者")
return victim

return None

class TransactionManager:
"""
事务管理器
管理事务的生命周期和并发控制
"""

def __init__(self, isolation_level: IsolationLevel = IsolationLevel.READ_COMMITTED):
"""
初始化事务管理器

Args:
isolation_level: 默认隔离级别
"""
self.isolation_level = isolation_level
self.concurrency_control = ConcurrencyControlManager()
self.transactions: Dict[str, Transaction] = {}
self.transaction_lock = threading.RLock()

# 监控线程
self.monitor_thread = threading.Thread(target=self._monitor_deadlocks, daemon=True)
self.monitor_thread.start()

# 统计数据
self.stats = {
"transactions_started": 0,
"transactions_committed": 0,
"transactions_aborted": 0,
"deadlocks_detected": 0,
"lock_timeouts": 0,
}

def begin_transaction(
self,
isolation_level: Optional[IsolationLevel] = None
) –> str:
"""
开始新事务

Args:
isolation_level: 隔离级别

Returns:
事务ID
"""
with self.transaction_lock:
# 生成事务ID
transaction_id = f"tx_{self.concurrency_control.get_next_timestamp()}_{uuid.uuid4().hex[:8]}"

# 确定隔离级别
level = isolation_level or self.isolation_level

# 创建事务
transaction = Transaction(
transaction_id=transaction_id,
timestamp=self.concurrency_control.get_next_timestamp(),
isolation_level=level
)

# 保存事务
self.transactions[transaction_id] = transaction
self.stats["transactions_started"] += 1

logger.info(f"开始事务: {transaction_id[:8]} (隔离级别: {level.value})")
return transaction_id

def read(
self,
transaction_id: str,
data_item: str,
use_mvcc: bool = False
) –> Any:
"""
读取数据

Args:
transaction_id: 事务ID
data_item: 数据项
use_mvcc: 是否使用MVCC

Returns:
读取的值
"""
with self.transaction_lock:
if transaction_id not in self.transactions:
raise ValueError(f"事务不存在: {transaction_id}")

transaction = self.transactions[transaction_id]

if transaction.status != TransactionStatus.ACTIVE:
raise ValueError(f"事务状态无效: {transaction.status}")

# 根据隔离级别选择并发控制策略
if transaction.isolation_level == IsolationLevel.SERIALIZABLE:
# 串行化：使用严格的锁
if not self.concurrency_control.lock_based_control(
transaction, data_item, "read"
):
self.stats["lock_timeouts"] += 1
raise TimeoutError(f"读取锁超时: {data_item}")

# 模拟读取数据
value = f"value_of_{data_item}_at_{time.time():.2f}"

elif use_mvcc or transaction.isolation_level == IsolationLevel.REPEATABLE_READ:
# 可重复读或使用MVCC
value = self.concurrency_control.mvcc_read(transaction, data_item)
if value is None:
value = f"value_of_{data_item}_at_{time.time():.2f}"

elif transaction.isolation_level == IsolationLevel.READ_COMMITTED:
# 读已提交：获取共享锁，读取后立即释放
if not self.concurrency_control.lock_based_control(
transaction, data_item, "read", timeout=1.0
):
self.stats["lock_timeouts"] += 1
raise TimeoutError(f"读取锁超时: {data_item}")

# 模拟读取数据
value = f"value_of_{data_item}_at_{time.time():.2f}"

# 立即释放读锁（读已提交特性）
self.concurrency_control.lock_manager.release_lock(
transaction_id, data_item, LockType.SHARED
)

else: # READ_UNCOMMITTED
# 读未提交：不获取锁，可能读取到未提交的数据
value = f"value_of_{data_item}_at_{time.time():.2f}"
transaction.read_set.add(data_item)

logger.debug(f"事务 {transaction_id[:8]} 读取 {data_item}: {value}")
return value

def write(
self,
transaction_id: str,
data_item: str,
value: Any,
use_optimistic: bool = False
) –> bool:
"""
写入数据

Args:
transaction_id: 事务ID
data_item: 数据项
value: 写入的值
use_optimistic: 是否使用乐观并发控制

Returns:
是否成功
"""
with self.transaction_lock:
if transaction_id not in self.transactions:
raise ValueError(f"事务不存在: {transaction_id}")

transaction = self.transactions[transaction_id]

if transaction.status != TransactionStatus.ACTIVE:
raise ValueError(f"事务状态无效: {transaction.status}")

success = False

if use_optimistic:
# 乐观并发控制
success = self.concurrency_control.optimistic_control(
transaction, data_item, "write", value
)

elif transaction.isolation_level == IsolationLevel.SERIALIZABLE:
# 串行化：使用排他锁
success = self.concurrency_control.lock_based_control(
transaction, data_item, "write"
)

else:
# 其他隔离级别：使用排他锁
success = self.concurrency_control.lock_based_control(
transaction, data_item, "write", timeout=2.0
)

if not success:
self.stats["lock_timeouts"] += 1
logger.warning(f"事务 {transaction_id[:8]} 写入失败: {data_item}")

return success

def commit(self, transaction_id: str) –> bool:
"""
提交事务

Args:
transaction_id: 事务ID

Returns:
是否成功提交
"""
with self.transaction_lock:
if transaction_id not in self.transactions:
raise ValueError(f"事务不存在: {transaction_id}")

transaction = self.transactions[transaction_id]

if transaction.status != TransactionStatus.ACTIVE:
raise ValueError(f"事务状态无效: {transaction.status}")

# 检查是否使用乐观并发控制
is_optimistic = len(transaction.snapshot) > 0

if is_optimistic:
# 乐观并发控制：验证并提交
success = self.concurrency_control.commit_optimistic_transaction(transaction)
else:
# 悲观并发控制：直接提交
success = True

if success:
transaction.status = TransactionStatus.COMMITTED

# 释放所有锁
self.concurrency_control.lock_manager.release_all_locks(transaction_id)

self.stats["transactions_committed"] += 1
logger.info(f"事务提交: {transaction_id[:8]}")
else:
# 验证失败，中止事务
self.abort(transaction_id)
logger.warning(f"事务验证失败: {transaction_id[:8]}")

return success

def abort(self, transaction_id: str) –> bool:
"""
中止事务

Args:
transaction_id: 事务ID

Returns:
是否成功中止
"""
with self.transaction_lock:
if transaction_id not in self.transactions:
raise ValueError(f"事务不存在: {transaction_id}")

transaction = self.transactions[transaction_id]

if transaction.status != TransactionStatus.ACTIVE:
raise ValueError(f"事务状态无效: {transaction.status}")

transaction.status = TransactionStatus.ABORTED

# 释放所有锁
self.concurrency_control.lock_manager.release_all_locks(transaction_id)

# 清理乐观控制的工作区
transaction.snapshot.clear()

self.stats["transactions_aborted"] += 1
logger.info(f"事务中止: {transaction_id[:8]}")

return True

def _monitor_deadlocks(self):
"""监控死锁"""
while True:
time.sleep(1.0) # 每秒检查一次

try:
victim = self.concurrency_control.detect_and_resolve_deadlock()

if victim:
self.stats["deadlocks_detected"] += 1

# 中止牺牲者事务
if victim in self.transactions:
transaction = self.transactions[victim]
if transaction.status == TransactionStatus.ACTIVE:
self.abort(victim)
logger.warning(f"因死锁中止事务: {victim[:8]}")
except Exception as e:
logger.error(f"死锁监控错误: {e}")

def get_transaction_status(self, transaction_id: str) –> Optional[Dict[str, Any]]:
"""
获取事务状态

Args:
transaction_id: 事务ID

Returns:
事务状态信息
"""
with self.transaction_lock:
if transaction_id not in self.transactions:
return None

transaction = self.transactions[transaction_id]

return {
"transaction_id": transaction.transaction_id,
"timestamp": transaction.timestamp,
"isolation_level": transaction.isolation_level.value,
"status": transaction.status.value,
"read_set": list(transaction.read_set),
"write_set": list(transaction.write_set),
"lock_requests": {
item: lock_type.value
for item, lock_type in transaction.lock_requests.items()
}
}

def get_system_status(self) –> Dict[str, Any]:
"""
获取系统状态

Returns:
系统状态信息
"""
with self.transaction_lock:
return {
"active_transactions": len([
t for t in self.transactions.values()
if t.status == TransactionStatus.ACTIVE
]),
"total_transactions": len(self.transactions),
"stats": self.stats.copy(),
"lock_status": self.concurrency_control.lock_manager.get_lock_status()
}

def clear_transactions(self):
"""清理完成的事务"""
with self.transaction_lock:
# 保留活跃事务，清理已完成的事务
self.transactions = {
tid: t for tid, t in self.transactions.items()
if t.status == TransactionStatus.ACTIVE
}

class Database:
"""
数据库类
提供简化的数据库操作接口
"""

def __init__(self, isolation_level: IsolationLevel = IsolationLevel.READ_COMMITTED):
"""
初始化数据库

Args:
isolation_level: 默认隔离级别
"""
self.transaction_manager = TransactionManager(isolation_level)
self.data_store: Dict[str, Any] = {}
self.data_lock = threading.RLock()

@contextmanager
def transaction(self, isolation_level: Optional[IsolationLevel] = None):
"""
事务上下文管理器

Args:
isolation_level: 隔离级别
"""
tx_id = None
try:
tx_id = self.transaction_manager.begin_transaction(isolation_level)
yield tx_id
self.transaction_manager.commit(tx_id)
except Exception as e:
if tx_id:
self.transaction_manager.abort(tx_id)
raise e

def get(self, key: str, transaction_id: Optional[str] = None) –> Any:
"""
获取数据

Args:
key: 键
transaction_id: 事务ID（如果为None，创建新事务）

Returns:
值
"""
if transaction_id:
return self.transaction_manager.read(transaction_id, key)
else:
# 自动事务
with self.transaction() as tx_id:
return self.transaction_manager.read(tx_id, key)

def set(self, key: str, value: Any, transaction_id: Optional[str] = None) –> bool:
"""
设置数据

Args:
key: 键
value: 值
transaction_id: 事务ID（如果为None，创建新事务）

Returns:
是否成功
"""
if transaction_id:
return self.transaction_manager.write(transaction_id, key, value)
else:
# 自动事务
with self.transaction() as tx_id:
return self.transaction_manager.write(tx_id, key, value)

def batch_operations(self, operations: List[Tuple[str, Any, str]]) –> bool:
"""
批量操作

Args:
operations: 操作列表，每个元素是(操作类型, 键, 值)
操作类型: "get" 或 "set"

Returns:
是否全部成功
"""
with self.transaction(IsolationLevel.SERIALIZABLE) as tx_id:
for op_type, key, value in operations:
if op_type == "get":
result = self.transaction_manager.read(tx_id, key)
logger.debug(f"批量读取: {key} = {result}")
elif op_type == "set":
success = self.transaction_manager.write(tx_id, key, value)
if not success:
return False
else:
raise ValueError(f"无效操作类型: {op_type}")
return True

def get_status(self) –> Dict[str, Any]:
"""
获取数据库状态

Returns:
状态信息
"""
return self.transaction_manager.get_system_status()

def simulate_concurrent_access():
"""
模拟并发访问场景
"""
print("=" * 80)
print("并发访问模拟")
print("=" * 80)

# 创建数据库
db = Database(isolation_level=IsolationLevel.READ_COMMITTED)

# 初始化数据
with db.transaction() as tx_id:
db.set("account_A", 1000, tx_id)
db.set("account_B", 1000, tx_id)

print("初始状态: account_A=1000, account_B=1000")

# 模拟转账操作
def transfer(from_account: str, to_account: str, amount: int, delay: float = 0):
"""转账函数"""
try:
with db.transaction(IsolationLevel.SERIALIZABLE) as tx_id:
# 读取余额
from_balance = int(db.get(from_account, tx_id))
to_balance = int(db.get(to_account, tx_id))

print(f"[{threading.current_thread().name}] "
f"读取: {from_account}={from_balance}, {to_account}={to_balance}")

# 模拟处理延迟
time.sleep(delay)

# 检查余额
if from_balance >= amount:
# 更新余额
db.set(from_account, from_balance – amount, tx_id)
db.set(to_account, to_balance + amount, tx_id)

print(f"[{threading.current_thread().name}] "
f"转账成功: {from_account} -> {to_account} 金额: {amount}")
return True
else:
print(f"[{threading.current_thread().name}] 余额不足")
return False
except Exception as e:
print(f"[{threading.current_thread().name}] 转账失败: {e}")
return False

# 创建并发转账线程
threads = []

# 线程1: A向B转账200
t1 = threading.Thread(
target=transfer,
args=("account_A", "account_B", 200, 0.5),
name="Thread-1"
)

# 线程2: B向A转账300
t2 = threading.Thread(
target=transfer,
args=("account_B", "account_A", 300, 0.3),
name="Thread-2"
)

# 线程3: A向B转账150（可能因死锁而失败）
t3 = threading.Thread(
target=transfer,
args=("account_A", "account_B", 150, 1.0),
name="Thread-3"
)

threads.extend([t1, t2, t3])

# 启动线程
for t in threads:
t.start()

# 等待线程完成
for t in threads:
t.join()

# 显示最终状态
final_a = db.get("account_A")
final_b = db.get("account_B")

print(f"\\n最终状态: account_A={final_a}, account_B={final_b}")

# 验证总额不变
total = int(final_a) + int(final_b)
print(f"总额验证: 1000+1000={total} ({'正确' if total == 2000 else '错误'})")

# 显示系统状态
status = db.get_status()
print(f"\\n系统状态:")
print(f" 活跃事务: {status['active_transactions']}")
print(f" 总事务数: {status['total_transactions']}")
print(f" 死锁检测次数: {status['stats']['deadlocks_detected']}")
print(f" 锁超时次数: {status['stats']['lock_timeouts']}")

def demonstrate_isolation_levels():
"""
演示不同隔离级别下的并发问题
"""
print("\\n" + "=" * 80)
print("隔离级别演示")
print("=" * 80)

isolation_levels = [
(IsolationLevel.READ_UNCOMMITTED, "脏读"),
(IsolationLevel.READ_COMMITTED, "不可重复读"),
(IsolationLevel.REPEATABLE_READ, "幻读"),
(IsolationLevel.SERIALIZABLE, "串行化"),
]

for isolation_level, problem in isolation_levels:
print(f"\\n隔离级别: {isolation_level.value}")
print(f"演示问题: {problem}")

db = Database(isolation_level=isolation_level)

# 初始化数据
with db.transaction() as tx_id:
db.set("counter", 0, tx_id)
db.set("data_1", "initial", tx_id)
db.set("data_2", "initial", tx_id)

# 创建两个事务
def transaction1():
"""事务1：读取并修改数据"""
try:
tx_id = db.transaction_manager.begin_transaction(isolation_level)

# 读取数据
val1 = db.get("data_1", tx_id)
print(f" 事务1读取 data_1: {val1}")

# 修改数据
time.sleep(0.2) # 给事务2时间读取
db.set("data_1", "modified_by_tx1", tx_id)
print(f" 事务1修改 data_1")

# 等待一会儿
time.sleep(0.5)

# 提交
db.transaction_manager.commit(tx_id)
print(f" 事务1提交")

except Exception as e:
print(f" 事务1失败: {e}")

def transaction2():
"""事务2：在不同时间点读取数据"""
try:
tx_id = db.transaction_manager.begin_transaction(isolation_level)

# 第一次读取
time.sleep(0.1)
val1 = db.get("data_1", tx_id)
print(f" 事务2第一次读取 data_1: {val1}")

# 第二次读取（可能在事务1提交后）
time.sleep(0.5)
val2 = db.get("data_1", tx_id)
print(f" 事务2第二次读取 data_1: {val2}")

# 检查是否可重复读
if val1 != val2:
print(f" ⚠️ 不可重复读: 第一次={val1}, 第二次={val2}")

# 尝试读取不存在的数据
try:
val3 = db.get("data_3", tx_id)
print(f" 事务2读取 data_3: {val3}")
except:
print(f" 事务2读取 data_3: 不存在")

# 提交
db.transaction_manager.commit(tx_id)
print(f" 事务2提交")

except Exception as e:
print(f" 事务2失败: {e}")

# 运行事务
t1 = threading.Thread(target=transaction1, name="Tx1")
t2 = threading.Thread(target=transaction2, name="Tx2")

t1.start()
t2.start()

t1.join()
t2.join()

print(f" 演示完成")

def performance_comparison():
"""
性能比较：悲观锁 vs 乐观锁 vs MVCC
"""
print("\\n" + "=" * 80)
print("性能比较")
print("=" * 80)

test_cases = [
("高冲突场景", 0.8), # 80%的操作会冲突
("中冲突场景", 0.3), # 30%的操作会冲突
("低冲突场景", 0.05), # 5%的操作会冲突
]

for scenario, conflict_rate in test_cases:
print(f"\\n{scenario} (冲突率: {conflict_rate*100:.0f}%)")

# 测试悲观锁
start_time = time.time()
db = Database(IsolationLevel.SERIALIZABLE)

with db.transaction() as tx_id:
for i in range(100):
key = f"item_{i % 10}" # 只有10个不同的key，增加冲突
db.set(key, i, tx_id)
if random.random() < conflict_rate:
time.sleep(0.001) # 模拟冲突延迟

pessimistic_time = time.time() – start_time

# 测试乐观锁
start_time = time.time()
db = Database(IsolationLevel.READ_COMMITTED)

# 使用乐观控制
tx_id = db.transaction_manager.begin_transaction(IsolationLevel.READ_COMMITTED)

try:
for i in range(100):
key = f"item_{i % 10}"
db.transaction_manager.write(tx_id, key, i, use_optimistic=True)
if random.random() < conflict_rate:
time.sleep(0.001)

success = db.transaction_manager.commit(tx_id)
if not success:
print(" 乐观锁: 事务验证失败")
except Exception as e:
print(f" 乐观锁错误: {e}")

optimistic_time = time.time() – start_time

print(f" 悲观锁时间: {pessimistic_time:.3f}秒")
print(f" 乐观锁时间: {optimistic_time:.3f}秒")
print(f" 性能提升: {(pessimistic_time – optimistic_time)/pessimistic_time*100:.1f}%")

def main():
"""
主函数：演示事务管理与并发控制
"""
print("事务管理与并发控制演示系统")
print("=" * 80)

# 1. 模拟并发访问
simulate_concurrent_access()

# 2. 演示隔离级别
demonstrate_isolation_levels()

# 3. 性能比较
performance_comparison()

# 4. 高级功能演示
print("\\n" + "=" * 80)
print("高级功能演示")
print("=" * 80)

# 创建数据库
db = Database(IsolationLevel.REPEATABLE_READ)

# 演示MVCC
print("\\nMVCC多版本并发控制演示:")

# 开始事务1
tx1 = db.transaction_manager.begin_transaction(IsolationLevel.REPEATABLE_READ)

# 事务1读取数据
print(f"事务1读取 data_x: {db.get('data_x', tx1)}")

# 开始事务2并修改数据
tx2 = db.transaction_manager.begin_transaction(IsolationLevel.REPEATABLE_READ)
db.set('data_x', 'modified_by_tx2', tx2)
db.transaction_manager.commit(tx2)
print(f"事务2修改并提交 data_x")

# 事务1再次读取（应该读取到旧版本）
print(f"事务1再次读取 data_x: {db.get('data_x', tx1)}")

db.transaction_manager.commit(tx1)

# 演示死锁检测
print("\\n死锁检测演示:")

db2 = Database(IsolationLevel.SERIALIZABLE)

def deadlock_task1():
tx = db2.transaction_manager.begin_transaction(IsolationLevel.SERIALIZABLE)
print(f"任务1获取锁A")
db2.set('A', 'value1', tx)
time.sleep(0.5)
print(f"任务1尝试获取锁B")
db2.set('B', 'value2', tx)
db2.transaction_manager.commit(tx)

def deadlock_task2():
tx = db2.transaction_manager.begin_transaction(IsolationLevel.SERIALIZABLE)
print(f"任务2获取锁B")
db2.set('B', 'value3', tx)
time.sleep(0.5)
print(f"任务2尝试获取锁A")
db2.set('A', 'value4', tx)
db2.transaction_manager.commit(tx)

t1 = threading.Thread(target=deadlock_task1, name="Deadlock-1")
t2 = threading.Thread(target=deadlock_task2, name="Deadlock-2")

t1.start()
t2.start()

t1.join()
t2.join()

print("\\n" + "=" * 80)
print("演示完成")
print("=" * 80)

if __name__ == "__main__":
main()

8. 分布式事务

8.1 两阶段提交（2PC）

两阶段提交是分布式事务中最经典的协议，包含两个阶段：

8.1.1 阶段一：准备阶段

8.1.2 阶段二：提交阶段

数学表示： 设

C

C

C是协调者，

P

=

P

1

,

P

2

,

.

.

.

,

P

n

P = {P_1, P_2, …, P_n}

P=P1​,P2​,…,Pn​是参与者集合

• 准备阶段：$\forall P_i\in P,C\to P_i:\text{PREPARE}$
• 提交阶段：如果$\forall P_i,P_i\to C:\text{YES}$
• 否则：$C\to P_i:\text{ABORT}$

8.2 三阶段提交（3PC）

三阶段提交解决了2PC的阻塞问题，增加了一个预提交阶段：

8.3 补偿事务（Saga模式）

Saga模式用于长时间运行的分布式事务，将大事务分解为一系列可补偿的小事务：

class SagaCoordinator:
"""Saga协调器"""

def execute_saga(self, transactions):
"""执行Saga事务"""
executed = []

try:
for i, transaction in enumerate(transactions):
# 执行事务
transaction.execute()
executed.append(transaction)

# 检查是否需要补偿
if not transaction.success:
# 反向补偿已执行的事务
for t in reversed(executed):
t.compensate()
return False

return True

except Exception as e:
# 发生异常，执行补偿
for t in reversed(executed):
t.compensate()
raise e

9. 现代并发控制技术

9.1 无锁数据结构

无锁数据结构通过原子操作（如CAS）实现并发安全，避免锁的开销：

import threading
import time
from typing import Any, Optional
import ctypes
import sys

class LockFreeStack:
"""无锁栈实现"""

class Node:
__slots__ = ['value', 'next']

def __init__(self, value):
self.value = value
self.next = None

def __init__(self):
self.head = None
self.counter = 0
self.lock = threading.Lock()

def push(self, value: Any):
"""压入元素"""
node = self.Node(value)

while True:
# 读取当前头节点
current_head = self.head

# 设置新节点的next为当前头节点
node.next = current_head

# 尝试原子地更新头节点
if self._compare_and_swap(self.head, current_head, node):
self.counter += 1
return

def pop(self) –> Optional[Any]:
"""弹出元素"""
while True:
# 读取当前头节点
current_head = self.head

if current_head is None:
return None

# 读取下一个节点
next_node = current_head.next

# 尝试原子地更新头节点
if self._compare_and_swap(self.head, current_head, next_node):
self.counter -= 1
return current_head.value

def _compare_and_swap(self, obj, expected, new):
"""比较并交换（模拟实现）"""
# 实际实现应使用原子操作
with self.lock:
if getattr(self, obj) == expected:
setattr(self, obj, new)
return True
return False

9.2 软件事务内存（STM）

软件事务内存将数据库的事务概念应用到内存操作：

class SoftwareTransactionalMemory:
"""软件事务内存"""

def __init__(self):
self.memory = {}
self.transactions = {}
self.lock = threading.RLock()

@contextmanager
def atomic(self):
"""原子事务上下文"""
tx_id = threading.get_ident()
snapshot = self.memory.copy()

try:
self.transactions[tx_id] = {
'snapshot': snapshot,
'writes': {},
'reads': set()
}

yield

# 提交事务
with self.lock:
# 验证读取的数据是否被修改
for key in self.transactions[tx_id]['reads']:
if key in self.memory and self.memory[key] != snapshot.get(key):
raise ValueError(f"数据冲突: {key}")