Celery任务中直接调用session.start_transaction()会失败，因MongoDB事务需独占ClientSession且不支持跨任务/线程复用；单节点mongod不支持事务，仅replica set或sharded cluster可用。
为什么 Celery 任务里直接用 session.start_transaction() 会失败
因为 Celery worker 默认复用 MongoDB 连接，而事务必须绑定到一个独占的 ClientSession 实例，且该 session 不能跨线程、跨协程、跨任务重用。常见错误现象是抛出 InvalidOperation: Session is ended 或 InvalidOperation: Transaction numbers are only allowed on a replica set member——后者说明你连的是单节点 mongod，压根不支持事务。
- 事务只在 replica set 或 sharded cluster 上可用，本地单机
mongod不行 - Celery 的
task.bind = True并不会自动给你配好 session；必须显式创建、使用、结束 - 不要在模块顶层或 task 外部缓存
session，它不是线程安全的 - 如果用了
pool=prefork（默认），每个子进程需独立初始化 client 和 session
如何在 @app.task 中正确开启并提交 MongoDB 事务
核心原则：session 生命周期必须严格限制在单个 task 执行范围内，且要兜底处理异常回滚。MongoDB Python Driver 要求事务内所有操作都通过同一个 session 对象调用。
- 用
with client.start_session() as session:确保自动清理，别手动调end_session() - 事务块内所有 collection 操作（
insert_one、update_many等）必须传入session=session - 捕获
PyMongoError后主动调session.abort_transaction()，否则可能卡住事务状态 - 避免在事务中做 HTTP 请求、调 Celery 子任务、写文件等非数据库操作——超时或失败会导致事务悬挂
@app.task(bind=True, autoretry_for=(PyMongoError,), retry_kwargs={'max_retries': 3})
def transfer_money(self, from_id, to_id, amount):
with mongo_client.start_session() as session:
try:
with session.start_transaction():
accounts = mongo_db.get_collection('accounts', session=session)
accounts.update_one({'_id': from_id}, {'$inc': {'balance': -amount}}, session=session)
accounts.update_one({'_id': to_id}, {'$inc': {'balance': amount}}, session=session)
except PyMongoError:
session.abort_transaction()
raise
并发任务下事务冲突和性能要注意什么
MongoDB 事务不是万能锁，它只对参与操作的文档加写锁，但事务执行时间越长，冲突概率越高，尤其在高并发转账类场景。真实问题是：你没法像 SQL 那样靠 SELECT ... FOR UPDATE 提前锁定。
- 事务最大默认时长是 60 秒（
transactionLifetimeLimitSeconds），超时自动 abort，Celery retry 可能重放失败逻辑 - 避免在事务里查大量数据或做复杂计算；先查再进事务，把只读操作挪到外面
- 如果两个任务同时修改同一文档，后提交者会报
WriteConflict，驱动会自动重试一次，但 Celery 默认不感知这个底层重试 - replica set 的 write concern（如
w: "majority"）会影响事务提交延迟，测试环境可临时设为w: 1加速验证
要不要在 Celery 中用 retry + 事务？怎么配才不重复扣款
要，但必须配合幂等设计。MongoDB 事务本身不解决重复执行问题，只保证单次执行的原子性。关键在：让重试后的第二次执行变成空操作。
立即学习“Python免费学习笔记（深入）”；
- 给每个任务加唯一
task_id或业务 id（如transfer_id），并在事务开头用find_one_and_update(..., upsert=True)记录“已执行”状态 - 不要依赖
self.request.id做幂等键——重试时 ID 不变，但你可能想让它变（比如加时间戳后缀） - 如果事务因网络中断没收到确认，但其实已提交，此时重试就会双花；所以状态记录动作本身也得在事务内完成
- 慎用
autoretry_for直接包整个事务函数——最好把事务逻辑抽成纯函数，外层 retry 控制是否重入
真正麻烦的是：事务 + 幂等 + 分布式锁 + 跨服务协调，这已经超出 MongoDB 单一能力范围。实际项目里，优先考虑用预占余额、异步对账、补偿任务这些更可控的模式。
