Python 慢函数的自动熔断装饰器

本篇文章向大家介绍《Python 慢函数的自动熔断装饰器》，主要包括，具有一定的参考价值，需要的朋友可以参考一下。

@functools.lru_cache不能当熔断器用，因其仅缓存结果、无超时控制、不统计失败率、无法响应临时故障；熔断需基于时间窗口的失败率/慢调用率、函数级状态隔离与硬超时机制。

为什么 @functools.lru_cache 不能当熔断用

缓存解决的是重复计算，不是超时或失败保护。哪怕函数卡死 10 秒，lru_cache 也照等不误，甚至把线程拖垮。熔断要的是“这次太慢，直接跳过”，不是“记得上次结果”。

• 缓存命中时快，但首次调用或未命中时完全不设防
• 没有超时控制，无法响应网络抖动、数据库锁表等临时故障
• 不统计失败率，没法判断服务是否真的不可用

tenacity 能做熔断但默认不启用

tenacity 本身是重试库，它的熔断能力藏在 wait=wait_none() + stop=stop_after_attempt(1) + retry=retry_if_exception_type(...) 组合里，但默认配置全是重试逻辑，直接拿来就用会误判为“重试装饰器”。

• 必须显式配 before_sleep 记录耗时，再靠外部状态机判断连续超时次数
• tenacity.Retrying 实例不能复用在多个函数上，否则状态混杂
• 超时检测得靠 timeout 参数（需配合 concurrent.futures 或 signal），而 Windows 不支持 signal.alarm

示例关键片段：

from tenacity import Retrying, retry_if_exception_type, stop_after_attempt
from concurrent.futures import ThreadPoolExecutor, TimeoutError

def make_circuit_breaker(timeout=2, max_failures=3):
    failures = []
    def before_sleep(retry_state):
        if retry_state.outcome.exception():
            failures.append(time.time())
            if len(failures) >= max_failures:
                # 这里触发熔断逻辑（比如 raise CircuitBreakerOpen）
                pass

手写轻量熔断装饰器的三个必要条件

真正能落地的熔断，不是看“失败几次”，而是看“最近 N 次里有多少次慢或失败”，且必须带时间窗口和状态隔离。缺一不可。

• 状态必须是函数粒度的：每个被装饰函数独享 _circuit_state 字典，不能共用全局变量
• 时间窗口要用 time.monotonic()，不用 time.time() —— 系统时间被 NTP 调整会导致窗口错乱
• 超时必须用 concurrent.futures.ThreadPoolExecutor.submit(...).result(timeout=...)，绕过 GIL 限制的同时保证硬超时

关键参数含义：failure_threshold 是失败率阈值（如 0.6），recovery_timeout 是熔断后多久尝试半开（如 60 秒），rolling_window 是滑动窗口秒数（如 60）

上线前必测的两个边界场景

多数人只测“连续失败”，却漏掉真实生产中最要命的两种情况：一种是“慢+快交替”，另一种是“刚恢复就来峰值”。不覆盖这两点，熔断器等于没装。

• 模拟 5 次调用：慢（3s）、快（0.1s）、慢（3s）、快（0.1s）、慢（3s）——检查是否因失败率未达标而未熔断
• 熔断打开后，在 recovery_timeout 刚过时并发发起 10 个请求——确认只有第一个进入半开，其余仍走 fallback，且半开失败立即回熔断

最容易被忽略的是 fallback 函数的异常处理：如果 fallback 自己抛错，整个装饰器会崩，必须用 try/except 包住 fallback 调用并兜底返回 None 或默认值。

本篇关于《Python 慢函数的自动熔断装饰器》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！