降级与故障恢复

更新 2026-10-05

我的学习 →

正在读取学习记录…

降级是在依赖不可用时，返回仍有明确用途、且注明限制的结果。库存实时查询失败时，可以返回最近一次库存并显示缓存年龄；缓存太旧则报告不可用。用户看到数据来自哪里，才能判断它是否足以支持下一步操作。

熔断是在连续失败后，暂时停止访问故障依赖；等待一段时间再探测，成功后恢复正常查询。降级决定失败时提供什么，熔断决定什么时候允许再调用，两种机制可以配合，但各自有状态和判断条件。

连续失败两次后暂停，再探测恢复

固定实时库存是8。服务在时刻0正常，时刻1和2失败，达到两次失败后暂停到时刻7。时刻3即使模拟服务可用，也暂不访问；时刻8探测成功，刷新缓存。随后再次失败，时刻20的缓存已超过10个时间单位。

程序使用传入的整数时间，便于复现等待边界；没有真正睡眠。每行打印实时或缓存来源、库存、缓存年龄和底层调用次数。

熔断暂停与缓存降级，缓存用于信息展示；写入业务仍需实时确认 — 图1：缓存用于信息展示；写入业务仍需实时确认。

保存完整程序并运行

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-fallback" -Force
cd "F:\学习\l4-fallback"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l4-fallback 文件夹，把下面完整代码保存为 fallback_demo.py，然后在PowerShell运行：

cd "F:\学习\l4-fallback"
ls
python -X utf8 fallback_demo.py

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-fallback" -Force
cd "F:\学习\l4-fallback"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""查询库存遇到故障时返回标明年龄的缓存，并探测恢复。"""


class Inventory:
    def __init__(self):
        self.failures = 0
        self.open_until = 0
        self.cache = None
        self.calls = 0

    def query(self, now, *, available):
        if now < self.open_until:
            return self.fallback(now)
        # 到达探测时间后允许一次调用；固定 available 模拟服务结果。
        self.calls += 1
        if available:
            self.failures = 0
            self.open_until = 0
            self.cache = (8, now)
            return {"stock": 8, "source": "live", "age": 0}
        self.failures += 1
        if self.failures >= 2:
            self.open_until = now + 5
        return self.fallback(now)

    def fallback(self, now):
        # 缓存超过10个时间单位就停止提供数值，保留明确不可用状态。
        if self.cache is None or now - self.cache[1] > 10:
            return {"stock": None, "source": "unavailable", "age": None}
        stock, saved = self.cache
        return {"stock": stock, "source": "cache", "age": now - saved}


def run_demo():
    service = Inventory()
    lines = []
    for now, available in ((0, True), (1, False), (2, False), (3, True),
                           (8, True), (9, False), (10, False), (20, False)):
        result = service.query(now, available=available)
        lines.append(f"时刻{now}: {result['source']}, 库存={result['stock']}, 年龄={result['age']}, 调用数={service.calls}")
    return lines


if __name__ == "__main__":
    print("\n".join(run_demo()))

完整固定输出如下：

时刻0: live, 库存=8, 年龄=0, 调用数=1
时刻1: cache, 库存=8, 年龄=1, 调用数=2
时刻2: cache, 库存=8, 年龄=2, 调用数=3
时刻3: cache, 库存=8, 年龄=3, 调用数=3
时刻8: live, 库存=8, 年龄=0, 调用数=4
时刻9: cache, 库存=8, 年龄=1, 调用数=5
时刻10: cache, 库存=8, 年龄=2, 调用数=6
时刻20: unavailable, 库存=None, 年龄=None, 调用数=7

时刻3为什么不增加调用数

时刻0来源 live 表示实时结果，缓存同时记录保存时间0。时刻1和2失败后返回 cache，年龄分别为1和2。时刻3仍在暂停窗口内，调用数保持3；程序直接使用年龄3的缓存，避免继续向故障服务施压。

时刻8到达探测时间，调用成功后清除失败数、解除暂停，并把缓存保存时间改为8。时刻20再次调用失败，缓存年龄已是12，超过可接受的10，返回 unavailable，库存为 None。库存未知与库存为0要分开，零库存会触发不同业务决策。

恢复探测怎样改变状态

open_until 保存暂停截止时间。now < open_until 期间直接降级；到达时间允许查询，成功就重置计数，失败达到阈值则再次暂停5个时间单位。单工作者示例相当于每次有一个调用参与探测。

多工作者系统需要限制恢复探测的并发数量，不能让所有等待请求同时冲向刚恢复的服务。常见做法是以条件更新或共享熔断器允许少量探测，其余请求继续降级。单纯用每个进程的独立失败计数，可能在大规模部署中失去预期效果。

哪些动作可以使用缓存

库存缓存可支持“最近看到的库存”展示，真正下单仍需实时校验或库存预留。审批权限、账户余额和支付结果也不能因为查询失败就直接用旧值执行动作。降级结果应带来源、时间和用途限制，让后续节点决定是否继续。

Agent 可以把缓存信息写进回答，但工具结果中的来源和年龄应保持可识别。若流程需要实时库存才能下单，收到缓存或不可用结果就转等待、重试或人工分支。让模型把缓存改写成“已确认有货”会破坏整个恢复机制。

失败分类和缓存有效期

网络超时、临时不可用可以累积进熔断统计；参数错误、无权限通常应直接修正请求或停止，不能统一算成服务故障。缓存年龄10是本例规则，实际值要根据数据变化频率和业务允许的偏差确定。

配套测试检查暂停期不调用、到期探测成功、探测失败再次暂停、无缓存和过期缓存。当前固定服务开关没有真实网络客户端、抖动和并发探测许可；接入服务后需要针对这些条件补验证。Resilience4j源码可进一步用于区分重试与熔断状态机。

改一处输入，核对机制变化

将 run_demo 中 (3, True) 改为 (7, True)，运行 python -X utf8 fallback_demo.py。时刻7恰好到达暂停截止值，结果变为live，调用数增为4；后续时刻8的调用数也相应增加。恢复时刻3并重跑，暂停期再次直接读取缓存。

运行配套测试

将下面测试保存为同一文件夹的 test_fallback_demo.py。它使用固定场景检查状态与业务结果，可以直接运行：

python -X utf8 -m unittest -v test_fallback_demo.py

完整测试：test_fallback_demo.py

import unittest
from fallback_demo import Inventory


class FallbackTests(unittest.TestCase):
    def test_open_skips_call_and_probe_recovers(self):
        # 暂停窗口不访问依赖，到期探测成功才清除失败计数。
        service = Inventory()
        service.query(0, available=True)
        service.query(1, available=False)
        service.query(2, available=False)
        self.assertEqual(service.query(3, available=True)["source"], "cache")
        self.assertEqual(service.calls, 3)
        self.assertEqual(service.query(7, available=True)["source"], "live")
        self.assertEqual(service.failures, 0)

    def test_stale_cache_is_unavailable(self):
        service = Inventory()
        service.query(0, available=True)
        self.assertEqual(service.query(11, available=False)["source"], "unavailable")

    def test_no_cache_is_unavailable(self):
        self.assertIsNone(Inventory().query(0, available=False)["stock"])

    def test_failed_probe_opens_again(self):
        service = Inventory()
        service.query(0, available=False)
        service.query(1, available=False)
        service.query(6, available=False)
        calls = service.calls
        service.query(7, available=True)
        self.assertEqual(service.calls, calls)


if __name__ == "__main__":
    unittest.main()

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“降级与故障恢复”，哪种做法更稳妥？

对应知识点：L4-fallback

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L4-fallback

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L4-fallback

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L4-fallback

你的回答