📚 课程目录第1章实现重试模式处理瞬时故障

1. 实现重试模式处理瞬时故障

任务说明

本案例要求实现一个重试模式组件，用于处理网络请求、数据库连接等场景中的瞬时故障。瞬时故障是指短暂存在、重试后可能成功的错误（如连接超时、临时性资源不可用），区别于永久性错误（如参数错误、权限不足）。完成目标：提供一个可配置重试次数、重试间隔和退避策略的重试器，能够区分瞬时故障与永久故障，并在重试耗尽后向上抛出最终异常。

核心实现

import time
import random
from typing import Callable, Type, Tuple, List, Any

class RetryableError(Exception):
    """标记可重试的瞬时故障基类"""
    pass

class ConnectionTimeoutError(RetryableError):
    """示例：连接超时（瞬时故障）"""
    pass

class DatabaseBusyError(RetryableError):
    """示例：数据库忙（瞬时故障）"""
    pass

class PermanentError(Exception):
    """永久性错误，不进行重试"""
    pass

class RetryConfig:
    def __init__(self, max_retries: int = 3, base_delay: float = 0.1,
                 max_delay: float = 2.0, exponential: bool = True,
                 jitter: bool = True):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.exponential = exponential
        self.jitter = jitter

class RetryExecutor:
    def __init__(self, config: RetryConfig = None):
        self.config = config or RetryConfig()

    def _compute_delay(self, attempt: int) -> float:
        """计算第 attempt 次重试前的等待时间（attempt 从 1 开始）"""
        if self.config.exponential:
            delay = self.config.base_delay * (2 ** (attempt - 1))
        else:
            delay = self.config.base_delay
        delay = min(delay, self.config.max_delay)
        if self.config.jitter:
            delay = random.uniform(0, delay)
        return delay

    def execute(self, func: Callable, *args, **kwargs) -> Any:
        """
        执行 func，遇 RetryableError 按配置重试；
        遇 PermanentError 或其他异常直接抛出，不重试。
        """
        attempt = 0
        while True:
            try:
                return func(*args, **kwargs)
            except RetryableError as e:
                if attempt >= self.config.max_retries:
                    raise  # 重试耗尽，抛出最后一次异常
                delay = self._compute_delay(attempt + 1)
                time.sleep(delay)
                attempt += 1
            except PermanentError:
                raise  # 永久错误，不重试
            # 其他未知异常：不重试，直接抛出
            # 如需对未知异常也重试，可在此扩展

# ---------- 使用示例 ----------
def unstable_network_call(url: str) -> str:
    """模拟一个可能抛出瞬时故障的函数"""
    # 模拟随机失败：前两次抛连接超时，第三次成功
    if not hasattr(unstable_network_call, "_call_count"):
        unstable_network_call._call_count = 0
    unstable_network_call._call_count += 1
    if unstable_network_call._call_count <= 2:
        raise ConnectionTimeoutError(f"连接 {url} 超时")
    return f"成功响应来自 {url}"

def invalid_request_call():
    """模拟永久性错误"""
    raise PermanentError("请求参数非法")

if __name__ == "__main__":
    # 配置：最多重试 3 次，指数退避，基础间隔 0.05 秒
    config = RetryConfig(max_retries=3, base_delay=0.05, max_delay=0.5)
    executor = RetryExecutor(config)

    # 场景1：瞬时故障，重试后成功
    result = executor.execute(unstable_network_call, "https://api.example.com/data")
    print(f"场景1结果: {result}")

    # 场景2：永久错误，不重试直接抛出
    try:
        executor.execute(invalid_request_call)
    except PermanentError as e:
        print(f"场景2捕获永久错误: {e}")

    # 场景3：重试耗尽后抛出最后一次异常
    def always_fail():
        raise DatabaseBusyError("数据库持续繁忙")

    try:
        executor.execute(always_fail)
    except DatabaseBusyError as e:
        print(f"场景3重试耗尽后抛出: {e}")

关键设计说明

故障分类：通过 RetryableError 基类标记瞬时故障，只有该类型异常触发重试；PermanentError 及未捕获的未知异常直接上抛，避免无效重试。
退避策略：默认采用指数退避（2^n 倍基础间隔），并叠加随机抖动（jitter）防止重试风暴；可通过 RetryConfig 切换为固定间隔或关闭抖动。
重试耗尽：达到 max_retries 后不再等待，直接抛出最后一次异常，由调用方决定后续处理。

落地与验证

运行方式：将上述代码保存为 retry_pattern.py，在 Python 3.6+ 环境直接执行 python retry_pattern.py。

验证方法：观察控制台输出。场景1应打印成功响应（因为第三次调用成功）；场景2应捕获 PermanentError 且不产生任何重试延迟；场景3应捕获 DatabaseBusyError，且从程序开始到抛出异常的总耗时约为 0.05+0.1+0.2=0.35 秒（含抖动，实际在 0~0.35 秒范围），证明重试了3次且每次间隔符合指数退避。

判定标准：三个场景输出与上述描述一致，且场景3的耗时明显大于单次调用耗时，说明重试机制生效；若将 max_retries 改为 0，场景1应立即抛出 ConnectionTimeoutError，验证重试次数配置生效。