深入解析事务基础与原子操作原理
title: 深入解析事务基础与原子操作原理
date: 2025/04/23 12:07:08
updated: 2025/04/23 12:07:08
author: cmdragon
excerpt:
事务是数据库操作的核心，需满足ACID特性：原子性、一致性、隔离性和持久性。事务隔离级别包括读未提交、读已提交、可重复读和串行化，适用于不同场景。嵌套事务通过上下文管理实现，支持回滚点以部分回滚。实战案例展示了订单系统中的事务处理，包括余额检查和支付失败回滚。常见错误如TransactionManagementError和死锁，需通过正确使用事务装饰器和重试机制解决。课后Quiz探讨了事务隔离和嵌套事务回滚的处理方法。运行环境配置包括依赖安装和FastAPI启动设置。
categories:
- • 后端开发
- • FastAPI
tags:
- • 事务处理
- • ACID特性
- • 嵌套事务
- • 回滚点
- • 订单系统
- • 事务隔离级别
- • 数据库操作
扫描二维码关注或者微信搜一搜：编程智域 前端至全栈交流与成长
探索数千个预构建的 AI 应用，开启你的下一个伟大创意
1. 事务基础与原子操作原理
1.1 事务的ACID特性
在数据库操作中，事务需要满足ACID特性：
- • 原子性（Atomicity）：操作要么全部成功，要么全部失败
- • 一致性（Consistency）：保持数据库的完整性约束
- • 隔离性（Isolation）：并发事务相互隔离
- • 持久性（Durability）：提交后永久保存
# 示例：银行转账的原子操作
from fastapi import APIRouter, Depends
from tortoise.transactions import atomic
router = APIRouter()
@router.post("/transfer")
@atomic() # 使用装饰器包裹事务范围
async def transfer_money(
from_account: str,
to_account: str,
amount: float
):
# 扣减转出账户
from_acc = await Account.get(number=from_account)
from_acc.balance -= amount
await from_acc.save()
# 增加转入账户
to_acc = await Account.get(number=to_account)
to_acc.balance += amount
await to_acc.save()
return {"message": "转账成功"}1.2 事务隔离级别对比
| 级别 | 脏读 | 不可重复读 | 幻读 | 适用场景 |
| 读未提交 | 可能 | 可能 | 可能 | 低并发场景 |
| 读已提交 | 禁止 | 可能 | 可能 | 默认级别 |
| 可重复读 | 禁止 | 禁止 | 可能 | 金融系统 |
| 串行化 | 禁止 | 禁止 | 禁止 | 高精度要求 |
2. 嵌套事务实现与回滚点
2.1 嵌套事务上下文管理
from tortoise.transactions import in_transaction
async def complex_operation():
async with in_transaction() as conn1: # 外层事务
await Model1.create(...)
try:
async with in_transaction(connection=conn1) as conn2: # 内层事务
await Model2.create(...)
await conn2.rollback() # 仅回滚内层操作
except Exception:
pass
await Model3.create(...) # 外层事务继续执行2.2 回滚点（Savepoint）使用
async def savepoint_demo():
async with in_transaction() as conn:
savepoint = await conn.savepoint() # 创建回滚点
try:
await Model.create(...)
if error_condition:
await savepoint.rollback() # 回滚到保存点
except Exception:
await savepoint.rollback()3. 完整实战案例：订单系统
from fastapi import Depends
from tortoise.contrib.fastapi import register_tortoise
from tortoise.models import Model
from tortoise import fields
# 数据模型定义
class User(Model):
id = fields.IntField(pk=True)
balance = fields.DecimalField(max_digits=10, decimal_places=2)
class Order(Model):
id = fields.UUIDField(pk=True)
user = fields.ForeignKeyField('models.User')
amount = fields.DecimalField(max_digits=10, decimal_places=2)
status = fields.CharField(max_length=20)
# 事务处理服务
class OrderService:
@staticmethod
@atomic()
async def create_order(user_id: int, amount: float):
user = await User.get(id=user_id)
# 检查余额
if user.balance < amount:
raise ValueError("余额不足")
# 扣减余额
user.balance -= amount
await user.save()
# 创建订单记录
order = await Order.create(
user=user,
amount=amount,
status="PENDING"
)
# 模拟第三方支付调用
if not await call_payment_gateway():
await OrderService.rollback_order(order.id)
raise Exception("支付失败")
return order
@staticmethod
@atomic()
async def rollback_order(order_id: str):
order = await Order.get(id=order_id)
user = await order.user
# 恢复用户余额
user.balance += order.amount
await user.save()
# 更新订单状态
order.status = "CANCELED"
await order.save()4. 常见报错解决方案
4.1 TransactionManagementError
错误现象：TransactionManagementError: Transaction not found for current context
解决方法：
- 1. 检查事务装饰器的使用范围
- 2. 确保异步函数正确使用async/await
- 3. 验证数据库连接配置是否正确
4.2 死锁检测
错误日志：Deadlock found when trying to get lock
处理方案：
from tortoise.exceptions import OperationalError
from fastapi import HTTPException
async def safe_transaction():
try:
async with in_transaction():
# 数据库操作
except OperationalError as e:
if "Deadlock" in str(e):
await asyncio.sleep(0.1) # 随机延迟后重试
return await safe_transaction()
else:
raise HTTPException(status_code=500, detail="数据库错误")5. 课后Quiz
5.1 事务隔离问题
问题：在可重复读隔离级别下，如何处理余额校验时的并发修改？
答案解析：
使用SELECT FOR UPDATE锁定记录：
async def update_balance(user_id: int):
async with in_transaction():
user = await User.select_for_update().get(id=user_id)
# 后续操作5.2 嵌套事务回滚
问题：当外层事务捕获内层事务的异常时，如何保证部分提交？
正确答案：
使用保存点机制：
async def nested_transaction():
async with in_transaction() as conn:
savepoint = await conn.savepoint()
try:
# 内层操作
except Exception:
await savepoint.rollback()
# 外层继续执行6. 运行环境配置
安装依赖：
pip install fastapi tortoise-orm uvicorn pydantic启动配置：
# main.py
from fastapi import FastAPI
app = FastAPI()
register_tortoise(
app,
db_url='sqlite://db.sqlite3',
modules={'models': ['models']},
generate_schemas=True,
add_exception_handlers=True
)启动命令：
uvicorn main:app --reload余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长，阅读完整的文章：深入解析事务基础与原子操作原理 | cmdragon's Blog
往期文章归档：
- • 掌握Tortoise-ORM高级异步查询技巧 | cmdragon's Blog
- • FastAPI与Tortoise-ORM实现关系型数据库关联 | cmdragon's Blog
- • Tortoise-ORM与FastAPI集成：异步模型定义与实践 | cmdragon's Blog
- • 异步编程与Tortoise-ORM框架 | cmdragon's Blog
- • FastAPI数据库集成与事务管理 | cmdragon's Blog
- • FastAPI与SQLAlchemy数据库集成 | cmdragon's Blog
- • FastAPI与SQLAlchemy数据库集成与CRUD操作 | cmdragon's Blog
- • FastAPI与SQLAlchemy同步数据库集成 | cmdragon's Blog
- • SQLAlchemy 核心概念与同步引擎配置详解 | cmdragon's Blog
- • FastAPI依赖注入性能优化策略 | cmdragon's Blog
- • FastAPI安全认证中的依赖组合 | cmdragon's Blog
- • FastAPI依赖注入系统及调试技巧 | cmdragon's Blog
- • FastAPI依赖覆盖与测试环境模拟 | cmdragon's Blog
- • FastAPI中的依赖注入与数据库事务管理 | cmdragon's Blog
- • FastAPI依赖注入实践：工厂模式与实例复用的优化策略 | cmdragon's Blog
- • FastAPI依赖注入：链式调用与多级参数传递 | cmdragon's Blog
- • FastAPI依赖注入：从基础概念到应用 | cmdragon's Blog
- • FastAPI中实现动态条件必填字段的实践 | cmdragon's Blog
- • FastAPI中Pydantic异步分布式唯一性校验 | cmdragon's Blog
- • 掌握FastAPI与Pydantic的跨字段验证技巧 | cmdragon's Blog
- • FastAPI中的Pydantic密码验证机制与实现 | cmdragon's Blog
- • 深入掌握FastAPI与OpenAPI规范的高级适配技巧 | cmdragon's Blog
- • Pydantic字段元数据指南：从基础到企业级文档增强 | cmdragon's Blog
- • Pydantic Schema生成指南：自定义JSON Schema | cmdragon's Blog
- • Pydantic递归模型深度校验36计：从无限嵌套到亿级数据的优化法则 | cmdragon's Blog
- • Pydantic异步校验器深：构建高并发验证系统 | cmdragon's Blog
- • Pydantic根校验器：构建跨字段验证系统 | cmdragon's Blog
- • Pydantic配置继承抽象基类模式 | cmdragon's Blog
- • Pydantic多态模型：用鉴别器构建类型安全的API接口 | cmdragon's Blog
- • FastAPI性能优化指南：参数解析与惰性加载 | cmdragon's Blog
- • FastAPI依赖注入：参数共享与逻辑复用 | cmdragon's Blog
- • FastAPI安全防护指南：构建坚不可摧的参数处理体系 | cmdragon's Blog
- • FastAPI复杂查询终极指南：告别if-else的现代化过滤架构 | cmdragon's Blog
- • FastAPI 核心机制：分页参数的实现与最佳实践 | cmdragon's Blog
- •
评论
发表评论