不用pydantic，它不处理运行时失败；优先选tenacity，但它只重试不熔断，真正熔断需用circuitbreaker或手写状态机。
熔断器该用 pydantic 还是 tenacity？
直接说结论：不用 pydantic，它不处理运行时失败；优先选 tenacity，但要注意它默认不带熔断状态持久化——它只重试，不“断”。真正做熔断得靠 circuitbreaker 或手撸状态机。如果你已经在用 fastapi 或 httpx，tenacity + 自定义 before/after 钩子能快速搭出带降级的重试逻辑，但别误以为重试 = 熔断。
circuitbreaker 的 failure_threshold 怎么设才不误杀？
这个值不是拍脑袋定的。它代表「连续失败多少次后跳闸」，设太小（比如 2）会导致网络抖动就熔断；设太大（比如 20）会让雪崩持续太久。实际要看依赖的 SLA 和你的容忍窗口：
- 对延迟敏感服务（如支付回调），建议从
failure_threshold=5起步，配合timeout=1.0 - 对吞吐敏感服务（如日志上报），可放宽到
failure_threshold=10，但必须加volume_threshold限制触发频率 - 务必配
reset_timeout=60（单位秒），否则熔断后永远不恢复
注意：circuitbreaker 的计数是全局共享的，多线程下没问题，但异步协程（async def）里要用 aiocircuitbreaker，原版不兼容 await。
HTTP 请求熔断时怎么返回兜底数据？
不能只抛异常，下游服务挂了你还让调用方等超时。要在 except 块里立即返回结构一致的 mock 数据：
立即学习“Python免费学习笔记（深入）”；
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
@circuit(failure_threshold=5, reset_timeout=60)
def fetch_user(user_id: str) -> dict:
resp = httpx.get(f"https://api.example.com/users/{user_id}")
resp.raise_for_status()
return resp.json()
<h1>熔断触发时走这里</h1><p>def fallback_fetch_user(user_id: str) -> dict:
return {"id": user_id, "name": "未知用户", "status": "offline"}
关键点：
- 兜底函数签名必须和原函数完全一致（包括参数名、类型注解、返回类型）
- 不要在 fallback 里再发 HTTP 或查 DB——它必须是纯内存计算
- 如果原函数有缓存（如
@lru_cache），fallback 不会自动缓存，要单独加
为什么本地测试总发现不了熔断逻辑失效？
因为多数人用 time.sleep() 模拟慢请求，但 circuitbreaker 默认只统计异常（Exception），不统计超时。你得显式 raise 错误：
# ❌ 错误：sleep 不触发熔断
def bad_slow_api():
time.sleep(3)
return {"ok": True}
<h1>✅ 正确：主动抛异常，或用 tenacity 的 retry_if_exception_type 控制</h1><p>def good_slow_api():
if random.random() > 0.8:
raise httpx.TimeoutException("simulated timeout")
return {"ok": True}
更隐蔽的问题是：单元测试里没清空熔断器状态。每次测前必须调用 circuitbreaker.CircuitBreakerMonitor.get().reset_all()，否则上一次失败会影响本次测试结果。
真实环境里，熔断状态跨进程不共享，K8s 多副本下每个实例独立判断——这不是缺陷，是设计使然；但意味着你得看聚合指标（比如 Prometheus 的 circuit_breaker_state{service="xxx"}），不能只盯单个 Pod 日志。
