重试策略详解

章节：入门篇第4章第2节
难度：⭐ 入门级
阅读时间：15-20 分钟
核心收获：深入理解重试策略的设计和实现

7.7 理论要点：重试不是万能药

7.7.1 重试的适用场景

可以重试的情况： - 网络波动导致的超时 - 服务端暂时过载（429） - 临时性的服务不可用 - 推送通道暂时拥堵

不可以重试的情况： - 认证失败（401/403）——Key 过期或权限不足，重试无用 - 数据源已下线 ——服务永久不可用 - Prompt 语法错误 ——代码问题，需修复 - 输入数据格式错误 ——需人工修正

7.7.2 重试的代价

7.7.3 重试策略设计原则

1. 有限重试：最多重试 2-3 次，避免无限循环
2. 退避等待：每次重试前等待一段时间，指数退避
3. 区分故障类型：临时故障才重试，永久故障立即停止
4. 保留现场：记录重试次数和原因，便于排查

7.8 实战案例：三种重试策略

7.8.1 策略1：固定间隔重试

适用场景：网络超时等随机性故障

第1次：立即执行
第2次：等待10秒后执行
第3次：等待10秒后执行

代码示例：

def fixed_interval_retry(func, max_retries=2, interval=10):
    for attempt in range(max_retries + 1):
        try:
            return func()
        except Timeout:
            if attempt < max_retries:
                time.sleep(interval)
                continue
            raise

优点：简单易实现
缺点：对限流服务不友好，可能加剧冲突

7.8.2 策略2：指数退避重试

适用场景：API 限流、服务过载

第1次：立即执行
第2次：等待 2^1 = 2 秒
第3次：等待 2^2 = 4 秒
第4次：等待 2^3 = 8 秒

代码示例：

def exponential_backoff_retry(func, max_retries=3, base_delay=1):
    for attempt in range(max_retries + 1):
        try:
            return func()
        except RateLimitError as e:
            if attempt < max_retries:
                wait_time = base_delay * (2 ** attempt)
                # 如果服务端返回 Retry-After，优先使用
                if hasattr(e, 'retry_after'):
                    wait_time = max(wait_time, e.retry_after)
                time.sleep(wait_time)
                continue
            raise

优点：给服务恢复时间，减少冲突
缺点：总耗时较长

7.8.3 策略3：混合策略

适用场景：复杂系统，多种故障类型

超时 → 固定间隔重试（10秒）
限流 → 指数退避 + Retry-After
认证失败 → 不重试，立即告警

代码示例：

def smart_retry(func, max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            return func()
        except Timeout:
            if attempt < max_retries:
                time.sleep(10)  # 固定间隔
                continue
            return "timeout"
        except RateLimitError as e:
            if attempt < max_retries:
                wait_time = parse_retry_after(e) or (2 ** attempt)
                time.sleep(wait_time)  # 指数退避
                continue
            return "rate_limited"
        except AuthError:
            return "auth_failed"  # 不重试
    return "failed"

7.9 超时设置

7.9.1 超时分级

7.9.2 超时配置示例

Python requests：

response = requests.get(
    url,
    timeout=(5, 10)  # (连接超时, 读取超时)
)

豆包 Prompt：

每个数据源的超时设置为10秒。
如果10秒内没有响应，立即标记为超时并继续处理其他来源。

7.10 熔断器模式

7.10.1 什么是熔断器

当某个服务连续失败达到阈值时，自动停止请求，给服务恢复时间。

三种状态： - 关闭（Closed）：正常请求，统计失败率 - 打开（Open）：快速失败，不发起请求 - 半开（Half-Open）：允许少量请求探测，如果成功则关闭熔断器

7.10.2 熔断器配置

7.10.3 熔断器代码示例

class CircuitBreaker:
    def __init__(self, failure_threshold=0.5, min_requests=10, 
                 open_duration=30, half_open_max=3):
        self.failure_threshold = failure_threshold
        self.min_requests = min_requests
        self.open_duration = open_duration
        self.half_open_max = half_open_max

    def call(self, func):
        if self.state == "open":
            if time.time() - self.opened_at < self.open_duration:
                return "circuit_open"
            else:
                self.state = "half_open"
                self.half_open_attempts = 0

        try:
            result = func()
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            return "failed"

7.11 图文说明

📌 待补充：建议绘制以下示意图 1. 重试策略对比图（固定间隔 vs 指数退避 vs 混合） 2. 熔断器状态流转图（Closed → Open → Half-Open） 3. 超时设置决策树 4. 错误处理流程图

7.12 常见错误

📝 版本迭代记录