Tenacity 熔断与回退

示例说明

重试解决偶发抖动，熔断解决持续故障：下游彻底挂掉时继续重试只是白烧资源。状态机三态转换——闭合时正常放行并计数失败，连续三次失败跳闸进入打开态直接短路走回退，冷却期过后半开态放一个试探请求探活。tenacity 的 after 钩子在每次重试结束后回调，是天然的记账时机。

技术要点

• 失败阈值触发跳闸短路调用
• 冷却期后半开试探恢复
• 回退值保证调用方总有结果

代码解读

CircuitBreaker 的 allow 判定三态：无打开时间戳即闭合放行，冷却已过则重置为半开放行，否则拒绝；record 按成败维护计数与跳闸。after 钩子从 retry_state 读出失败结果记账。get_rates 是统一入口：熔断打开或调用失败都返回缓存汇率与来源标签，调用方永远有可用结果。

使用建议

阈值与冷却按下游恢复周期调。半开态的试探请求应限流防洪泛。回退数据的时效要标注，避免陈旧值被当实时值使用。

代码示例

import random
import time

from tenacity import RetryCallState, retry, stop_after_attempt, wait_fixed


class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, cooldown: float = 5.0):
        self.failure_threshold = failure_threshold
        self.cooldown = cooldown
        self.failures = 0
        self.opened_at: float | None = None

    def allow(self) -> bool:
        if self.opened_at is None:
            return True
        if time.monotonic() - self.opened_at >= self.cooldown:
            self.opened_at = None  # 半开：放行试探请求
            self.failures = 0
            return True
        return False

    def record(self, success: bool) -> None:
        if success:
            self.failures = 0
            self.opened_at = None
        else:
            self.failures += 1
            if self.failures >= self.failure_threshold:
                self.opened_at = time.monotonic()


breaker = CircuitBreaker(failure_threshold=3, cooldown=5)
FALLBACK_RATES = {'usd': 1.0, 'eur': 0.92}


def after_attempts(retry_state: RetryCallState) -> None:
    breaker.record(retry_state.outcome.failed)


@retry(stop=stop_after_attempt(2), wait=wait_fixed(0.1),
       after=after_attempts, reraise=True)
def call_rate_api() -> dict:
    if random.random() < 0.6:
        raise ConnectionError('rates service flaky')
    return {'usd': 1.0, 'eur': 0.91}


def get_rates() -> tuple[dict, str]:
    if not breaker.allow():
        return FALLBACK_RATES, 'fallback-open'
    try:
        return call_rate_api(), 'live'
    except ConnectionError:
        return FALLBACK_RATES, 'fallback-error'


for _ in range(8):
    rates, source = get_rates()
    print('source:', source, 'usd:', rates['usd'])