大家好我是豆子今天主要针对一致性防重中的锁机制方案做一个复盘欢迎大家讨论和指出不足之处需求背景一次问数请求需要持续管理执行权QueryData 是基于 SQLBot 二次开发的校园智能问数系统。用户输入自然语言问题后系统检索表结构与业务知识调用模型生成 SQL再执行查询并展示结果。查询失败时还可能进入有限次数的自动修复。这条链路的特点是执行时间不确定而且客户端连接与后台任务的生命周期不一致。例如用户提交“查询当前仍有挂科的学生”模型还在生成 SQL浏览器就断网了。随后用户重新提交或者在另一个标签页发送新问题。此时原任务可能仍在后台执行系统需要避免重复模型调用和同一会话的并发推进。因此我们设计了两种锁锁解决的问题请求锁同一逻辑操作因重试重复到达避免重复执行会话锁同一会话中的不同操作并发到达避免上下文和执行顺序混乱但仅仅获取锁还不够。锁需要设置过期时间防止进程崩溃后永久占用同时长任务又可能超过这个有效期。我们最终采用了有限 TTL、周期续期、执行权检查和任务完成后的安全释放。锁续期的目的是在任务仍合法执行时维持它的执行资格而不是无条件让锁一直存在。2. 上下文传递选择 ContextVar并设计 QuestionLease请求上下文不能保存在普通全局变量中FastAPI 的多个异步请求可以在同一个事件循环线程中交替执行。如果用普通全局变量保存“当前租约”请求 A 在等待期间请求 B 就可能覆盖它。A 恢复后读取到 B 的租约造成请求之间串值。局部变量逐层传参可以避免这个问题但租约需要经过接口、记录创建、模型任务初始化等多个环节。我们选择使用ContextVar将租约关联到当前请求的执行上下文减少层层传参。QuestionLease 管理一次执行的生命周期QuestionLease将本次执行需要的身份、锁和管理方法组织在一起内容作用keys、token记录持有的锁及其所有者身份started、deadline限制总运行时间判断当前租约有效期state、closed标记执行资格和关闭状态renewal保存续期协程的 Taskobserver保存观察后台任务的 Taskon_completed登记结果持久化与缓存回调请求 ID、指纹、记录 ID关联逻辑操作和数据库记录其中request_id与token分工不同request_id由前端生成标识一次逻辑操作网络重发时复用。token在后端创建租约时生成标识这一次锁持有者。Redis 锁的 Key 包含请求 IDValue 保存 token。释放和续期时依据 token 校验所有权。租约如何传递到模型线程模型任务初始化时从当前上下文取得租约self.execution_lease active_question_lease.get()随后模型任务被提交到线程池。工作线程通过self.execution_lease检查执行权。因此事件循环侧和模型工作线程侧持有的是同一个租约对象的引用没有复制两份对象进行同步也不依赖 ContextVar 自动跨线程传播。共享引用并不代表任意操作都自动线程安全。当前设计主要让事件循环侧负责异步 Redis、续期和收尾让模型线程在关键步骤检查本地执行资格。3. 整体链路续期协程与观察协程如何协作接口准入与任务启动一次新请求的主要链路是权限检查、请求指纹与回放检查 ↓ 获取请求锁启动续期 ↓ ContextVar 关联租约 ↓ 检查数据库原请求结果 ↓ 获取会话锁将两把锁统一纳入续期 ↓ 绑定请求信息与完成回调 ↓ 模型任务取得租约引用 ↓ 提交线程池工作得到 Future ↓ 启动观察协程重复请求如果已经有完成结果直接回放不重新启动模型。确实需要执行的新操作才竞争会话锁。续期协程负责维持资格获取请求锁后QuestionGate.acquire()启动续期任务lease.renewal asyncio.create_task(lease._renew())取得会话锁时acquire_chat()将会话锁加入租约立即统一续期再重新启动周期续期任务。当前默认配置是锁有效期60 秒。续期间隔20 秒。最大运行时长600 秒。续期协程周期调用renew_once()校验锁仍属于当前 token再重新设置 TTL。续期失败后租约进入LOST模型任务在后续检查点停止推进。这些时间是当前默认配置实际应结合模型耗时和调度延迟调整。观察协程负责等待真实任务结束模型任务提交线程池后得到一个Future。它表示工作是否完成以及结果或异常。随后调用lease.follow(future, failed)follow()创建观察协程其核心等待方式是await asyncio.shield(asyncio.wrap_future(future))wrap_future()将线程池 Future 适配成可以异步等待的对象。等待期间观察协程让出事件循环续期和其他请求仍可以运行。shield()限制取消向被等待 Future 传播但不能强制终止工作线程。为什么不能在接口返回或 SSE 断开时释放锁接口返回流式响应只表示开始向客户端传输SSE 断开也只表示输出通道中断。两者都不能证明模型线程结束。因此后台任务启动后租约的收尾责任交给观察协程真实 Future 完成 ↓ 判断成功或失败 ↓ 租约有效时持久化最终状态与结果 ↓ 写入回放缓存 ↓ 停止续期 ↓ 安全释放锁没有启动后台工作时由接口上下文直接清理已经启动观察者时则由观察者负责收尾。续期协程和观察协程都运行在事件循环中职责不同续期协程维持执行资格观察协程等待工作结束并收尾。4. 安全性与可靠性设计原子获取与所有者安全释放获取锁使用SET key token NX PX ttl释放锁则通过 Lua原子比较 token 后再删除if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) end return 0这样旧任务即使在锁过期后恢复也不会误删新执行者的锁。分开执行GET和DEL会留下竞争窗口因此不能替代这段脚本。两把锁全部校验通过才统一续期for i, key in ipairs(KEYS) do if redis.call(get, key) ~ ARGV[1] then return 0 end end for i, key in ipairs(KEYS) do redis.call(pexpire, key, ARGV[2]) end return 1任意一把锁失去所有权就不续期任何一把避免任务只持有部分锁却继续执行。本地有效期与运行上限租约使用time.monotonic()计算时间避免系统时钟调整影响判断deadline判断当前租约是否过期。started判断是否达到总运行时长上限。LOST和closed阻止失效任务继续推进。续期回复到达后还会再次检查本地资格避免迟到的成功回复重新激活已经失效的租约。执行权检查是合作式的它能阻止后续步骤不能立即终止已经发出的模型调用。持久化结果与锁状态分离数据库保存请求状态和结果Redis 保存锁与短期回放缓存。chat_record对(chat_id, request_id)设置唯一约束并保存请求指纹防止同一 ID 被错误用于不同内容。锁消失不代表任务成功。只有数据库终态结果可以回放遗留的非终态记录返回结果不可用不直接重新执行。完成时先提交数据库再缓存结果。因此缓存写入失败不会抹掉已经持久化的完成结果。