SQLAlchemy 的 commit() 本身不导致死锁，死锁源于多个事务竞争同一行锁并形成循环等待；常见于并发修改相同记录且未统一加锁顺序或锁持有时间过长。
SQLAlchemy commit() 为什么会触发死锁
死锁不是 commit() 自身导致的，而是多个事务在竞争同一组行级锁时，彼此等待对方释放锁形成的循环依赖。Flask 中常见场景是并发请求同时修改同一张表的相同记录（比如库存扣减、计数器更新），且事务未按统一顺序加锁或持有锁时间过长。
关键点在于：SQLAlchemy 的 commit() 是事务提交的终点，但锁实际在执行 UPDATE/SELECT FOR UPDATE 等语句时就已获取；如果业务逻辑中 commit() 前做了大量非数据库操作（如 HTTP 调用、文件读写、复杂计算），锁就会被长时间占用，大幅提升死锁概率。
如何缩短锁持有时间并避免竞争热点
核心策略是让事务“快进快出”——尽早执行写操作，尽快 commit()，不把锁拖到业务逻辑后半段。
- 把所有数据库变更（
session.add()、session.query().update()、session.execute(text("UPDATE ..."))）集中在事务开头，避免边查边改、反复查询再决定怎么改 - 禁用自动刷新（
autoflush=False）并手动控制刷新时机，防止隐式flush()在你不预期的位置触发 SQL 执行 - 对高频更新字段（如
order_status、stock_count），改用原子 SQL 操作，绕过 ORM 全量加载+修改流程：session.execute(text("UPDATE orders SET status = :new WHERE id = :id AND status = :old"), {"new": "shipped", "old": "pending", "id": order_id}) - 避免在事务中调用外部服务；必须调用时，先
commit()再发请求，或改用最终一致性方案（如发消息队列）
捕获死锁并自动重试的实用写法
MySQL 和 PostgreSQL 都会在死锁时抛出特定异常（MySQL 是 OperationalError 且 orig.args[0] == 1213，PostgreSQL 是 DeadlockDetected），不能靠捕获通用异常来重试。
立即学习“Python免费学习笔记（深入）”；
推荐用轻量装饰器封装重试逻辑，注意限制最大重试次数（通常 3 次足够）：
from sqlalchemy.exc import OperationalError, DeadlockDetected
import time
<p>def retry_on_deadlock(max_retries=3, delay=0.1):
def decorator(f):
def wrapped(*args, <strong>kwargs):
for i in range(max_retries):
try:
return f(*args, *<em>kwargs)
except (OperationalError, DeadlockDetected) as e:
if i == max_retries - 1:
raise
if "1213" in str(e) or "deadlock detected" in str(e).lower():
time.sleep(delay </em> (2 </strong> i)) # 指数退避
continue
raise
return None
return wrapped
return decorator</p><p>@retry_on_deadlock()
def update_stock(order_id):
session.execute(text("UPDATE products SET stock = stock - 1 WHERE id = :id"), {"id": order_id})
session.commit() # 注意：这里 commit 必须在重试范围内
⚠️ 特别注意：重试只适用于幂等操作；如果函数里混了非数据库副作用（如发邮件、写日志），重试会导致重复执行。
Flask 中 session 管理不当引发的隐式死锁风险
很多人把 db.session 当成线程安全对象直接在视图里复用，但在多线程/协程环境下（如 gevent 或 Uvicorn + Flask），未及时 remove() 会导致 session 泄漏、连接复用混乱，间接放大锁冲突。
务必确保每个请求生命周期内 session 正确绑定与清理：
- 使用
@app.teardown_appcontext或@bp.teardown_appcontext注册清理钩子：@app.teardown_appcontext def shutdown_session(exception=None): if exception and db.session.is_active: db.session.rollback() db.session.remove() - 避免在全局作用域或模块级定义
session = db.session；始终通过db.session获取当前上下文 session - 不用
scoped_session手动管理 scope，除非你明确需要跨请求共享 session（通常不该这么做）
死锁真正的难点不在识别，而在于它往往只在压测或上线后突发，且难以复现。最有效的防御是：用原子 SQL 替代 ORM 更新、强制事务边界清晰、拒绝在事务里做任何 IO。其他优化都是补救，这三条才是根因控制。
