“半开状态”是断路器模式中上次失败后冷却期满、由下一个请求试探的瞬时状态,依据单调时钟判断冷却是否结束,成功则闭合、失败则断开,非定时切换且不可持久。

什么是“半开状态”?先说清楚判断依据
断路器模式里的“半开状态”,不是靠定时器自动切的,而是上一次失败后,等一段冷却时间,再让下一个请求去试探——成功就切回闭合,失败就切回断开。time.time() 或 time.monotonic() 是最常用的时钟源,但用错会出问题:比如系统时间被 NTP 回拨,time.time() 可能倒退,导致冷却提前结束;必须用 time.monotonic() 才可靠。
怎么实现冷却时间检查?别硬写 if-else 堆逻辑
核心是维护两个字段:self._last_failure_time(上次失败时间戳)和 self._timeout(冷却秒数)。每次尝试前只做一件事:比较当前单调时间与上次失败时间的差值是否 ≥ self._timeout。
- 不要在每次调用时都重算状态,只在进入
call()或attempt()时动态判断 - 不要把“半开”存成独立状态变量,它只是“上次失败过且冷却已过”的瞬时结果
- 如果用
threading.Lock保护状态,锁粒度要小——只锁更新_last_failure_time和_state的那几行,别包住整个业务调用
if time.monotonic() - self._last_failure_time >= self._timeout:
return "half_open"并发下怎么避免多个请求同时闯入半开?用原子标记打桩
多个线程/协程几乎同时到达冷却终点,不加控制就会一起发请求,失去断路器意义。不能靠“先查后设”这种竞态操作,得用原子写入。
- Python 标准做法是用
threading.Event或threading.Lock配合标志位self._half_open_entered - 更轻量的是用
self._half_open_entered = True+try/except捕获RuntimeError(如果用setdefault风格的字典操作) - 推荐用
threading.Once的语义模拟:首次成功设置self._half_open_granted = True返回 True,其余返回 False
一旦某个请求拿到“半开入场券”,其他请求就得立刻拒绝或排队,不能等它返回再决定。
立即学习“Python免费学习笔记(深入)”;
为什么测试半开逻辑总翻车?漏掉了失败回调的触发时机
半开状态本身不持久,它的生命周期完全依赖“试探请求”的结果。很多人只测了“冷却到了→进半开”,却没验证“试探失败→立刻回断开”或“试探成功→切闭合并清空失败时间”。
- 必须在试探请求的
except分支里更新self._last_failure_time,否则下次还是半开 - 必须在试探成功的
else分支里重置self._last_failure_time = 0或删掉该字段,否则下次仍可能误判 - 异步场景下(如
asyncio),await后的回调必须在同一个任务上下文完成状态更新,不能丢给后台 task
半开不是状态机里的一个稳态,它是一次性门票——用完即焚,设计时得按这个直觉来组织逻辑。










