本文目录导读：
在Python中保障事务安全（Transaction Safety）的核心目标是在数据操作过程中保证 原子性（Atomicity）、一致性（Consistency）、隔离性（Isolation） 和 持久性（Durability），即ACID特性，以下是几个典型的案例和实现方案：
数据库事务（以SQLAlchemy为例）
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine('postgresql://user:pass@localhost/db')
Session = sessionmaker(bind=engine)
session = Session()
try:
# 开启事务
session.begin()
# 业务操作1：扣减账户A余额
account_a = session.query(Account).filter_by(id=1).with_for_update().one()
account_a.balance -= 100
# 业务操作2：增加账户B余额
account_b = session.query(Account).filter_by(id=2).with_for_update().one()
account_b.balance += 100
# 提交事务
session.commit()
except Exception as e:
# 回滚事务
session.rollback()
raise e
finally:
session.close()
关键点：
with_for_update()添加行级锁防止并发冲突try/except确保异常时自动回滚commit()确保所有操作原子性提交
Redis事务（使用Pipeline）
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
def transfer_money(from_account, to_account, amount):
pipe = r.pipeline(transaction=True)
try:
# 监控关键值
pipe.watch(from_account, to_account)
# 获取当前余额
from_balance = int(pipe.get(from_account) or 0)
to_balance = int(pipe.get(to_account) or 0)
if from_balance < amount:
raise ValueError("Insufficient funds")
# 开始事务
pipe.multi()
pipe.decrby(from_account, amount)
pipe.incrby(to_account, amount)
pipe.execute()
print("Transaction successful")
except redis.WatchError:
# 如果监控的值被修改，重试
transfer_money(from_account, to_account, amount)
finally:
pipe.reset()
关键点：
WATCH实现乐观锁MULTI/EXEC保证命令的原子性执行WatchError处理并发冲突重试
文件系统事务（使用临时文件）
import os
import tempfile
import shutil
def safe_update_config(config_path, new_content):
# 创建临时文件
temp_fd, temp_path = tempfile.mkstemp(dir=os.path.dirname(config_path))
try:
with os.fdopen(temp_fd, 'w') as temp_file:
# 写入新内容
temp_file.write(new_content)
temp_file.flush()
os.fsync(temp_file.fileno()) # 确保数据写入磁盘
# 原子替换文件
os.replace(temp_path, config_path)
except Exception:
# 清理临时文件
if os.path.exists(temp_path):
os.remove(temp_path)
raise
关键点：
- 先写入临时文件
os.replace()原子替换（POSIX系统保证）- 异常时删除临时文件保持一致性
分布式事务（两阶段提交示例）
from contextlib import contextmanager
@contextmanager
def distributed_transaction(services):
"""简化版两阶段提交"""
prepared = []
try:
# Phase 1: 准备阶段
for service in services:
if not service.prepare():
raise Exception(f"Prepare failed for {service}")
prepared.append(service)
# Phase 2: 提交阶段
for service in prepared:
service.commit()
except Exception:
# 回滚所有已准备的服务
for service in prepared:
try:
service.rollback()
except Exception:
# 记录回滚失败日志，需要人工干预
log.error(f"Rollback failed for {service}")
raise
关键点：
- 两阶段提交（2PC）确保分布式一致性
- 准备阶段预留资源
- 失败时回滚所有参与者
Python对象状态事务（Memento模式）
class TransactionalObject:
def __init__(self):
self._state = {}
self._checkpoint = None
def begin_transaction(self):
# 保存当前状态的快照
self._checkpoint = self._state.copy()
def commit(self):
# 清除checkpoint
self._checkpoint = None
def rollback(self):
if self._checkpoint is not None:
self._state = self._checkpoint
self._checkpoint = None
def update(self, key, value):
self._state[key] = value
# 使用示例
obj = TransactionalObject()
obj.begin_transaction()
obj.update("name", "Alice")
obj.update("age", 30)
# 如果操作失败，可以回滚
if error_condition:
obj.rollback() # 状态恢复
else:
obj.commit() # 确认修改
- 使用标准库：尽量使用数据库、消息队列等成熟中间件的事务支持
- 锁策略选择：
- 悲观锁：适合高冲突场景
- 乐观锁：适合低冲突场景
- 幂等性设计：保证重复操作的幂等性，支持重试机制
- 日志记录：记录事务操作日志，便于审计和恢复
- 超时处理：设置合理的超时时间，防止死锁
- 测试覆盖：包括并发测试、异常测试和边界测试
选择哪种方案取决于你的具体场景：单机数据库操作推荐ORM事务，分布式系统可能需要分布式事务框架（如Saga模式），而内存对象状态管理可以用备忘录模式。
