4、重试策略的代码实现：Python 通用重试装饰器与异步框架集成

代码落地，才是重试策略真正发挥作用的地方。这一章，我带你手写一个 Python 通用重试装饰器，再把它和 asyncio 结合起来。说白了，就是让你在同步和异步代码里都能优雅地重试。

4.1 为什么需要通用重试装饰器？

你想想看，如果每个需要重试的函数都写一遍 try-except-while 循环，代码会变成什么样？我见过一个项目，光重试逻辑就散落在 20 多个文件里，改个退避策略得翻遍整个仓库。嗯，这显然不是好做法。

装饰器模式可以把重试逻辑集中管理。你只需要在函数头上加一行 @retry，剩下的交给装饰器处理。这样既干净，又容易维护。

核心目标：一个装饰器，支持配置重试次数、退避策略、可重试异常类型，并且能用在同步和异步函数上。

4.2 同步版重试装饰器实现

我们先从同步版本开始。我个人习惯先写一个 RetryConfig 数据类，把配置参数集中管理。

import time
import functools
from typing import Tuple, Type, Callable, Optional
from dataclasses import dataclass, field

@dataclass
class RetryConfig:
    max_retries: int = 3
    base_delay: float = 1.0
    max_delay: float = 60.0
    backoff_factor: float = 2.0
    retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,)
    jitter: bool = True

def retry(config: Optional[RetryConfig] = None):
    if config is None:
        config = RetryConfig()
    
    def decorator(func: Callable):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(config.max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except config.retryable_exceptions as e:
                    last_exception = e
                    if attempt < config.max_retries:
                        delay = min(
                            config.base_delay * (config.backoff_factor ** attempt),
                            config.max_delay
                        )
                        if config.jitter:
                            import random
                            delay = delay * (0.5 + random.random())
                        time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

这段代码的核心逻辑很简单：

  • 每次调用函数，如果抛出可重试异常，就计算退避延迟。
  • 延迟时间按指数增长，但不超过 max_delay。
  • 如果开启了 jitter，延迟会随机抖动，防止惊群效应。
  • 重试次数用完后，抛出最后一次捕获的异常。

使用示例：

@retry(RetryConfig(max_retries=5, base_delay=0.5))
def fetch_data(url: str):
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()

4.3 配置化重试策略

实际项目中，不同接口的重试需求差别很大。比如：

  • 调用第三方支付接口，最多重试 3 次，间隔 1 秒。
  • 查询数据库，重试 5 次，间隔 0.1 秒，只重试连接超时异常。
  • 调用内部微服务，重试 2 次，间隔 5 秒，所有异常都重试。

我建议把配置外移到 YAML 或环境变量里。这样改策略不用改代码，运维同学也能直接调。

# retry_config.yaml
payment_api:
  max_retries: 3
  base_delay: 1.0
  backoff_factor: 2.0
  retryable_exceptions: ["requests.exceptions.Timeout", "requests.exceptions.ConnectionError"]

db_query:
  max_retries: 5
  base_delay: 0.1
  backoff_factor: 1.5
  retryable_exceptions: ["pymongo.errors.AutoReconnect"]

加载配置的代码也很简单：

import yaml
from typing import Dict

def load_retry_config(path: str) -> Dict[str, RetryConfig]:
    with open(path) as f:
        raw = yaml.safe_load(f)
    configs = {}
    for name, cfg in raw.items():
        configs[name] = RetryConfig(
            max_retries=cfg.get("max_retries", 3),
            base_delay=cfg.get("base_delay", 1.0),
            backoff_factor=cfg.get("backoff_factor", 2.0),
            retryable_exceptions=tuple(
                eval(exc) for exc in cfg.get("retryable_exceptions", ["Exception"])
            )
        )
    return configs

注意：用 eval 加载异常类有安全风险。生产环境建议用异常类注册表或字符串映射，避免执行任意代码。

4.4 结合 asyncio 的异步重试

异步编程越来越普遍。我最近维护的一个数据管道，全是 async/await 写的。同步重试装饰器在异步函数上会阻塞事件循环，这是大忌。

异步重试的核心区别在于：用 asyncio.sleep 代替 time.sleep，用 await 调用异步函数。

import asyncio
import functools
from typing import Any

def async_retry(config: Optional[RetryConfig] = None):
    if config is None:
        config = RetryConfig()
    
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(config.max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except config.retryable_exceptions as e:
                    last_exception = e
                    if attempt < config.max_retries:
                        delay = min(
                            config.base_delay * (config.backoff_factor ** attempt),
                            config.max_delay
                        )
                        if config.jitter:
                            import random
                            delay = delay * (0.5 + random.random())
                        await asyncio.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

用法和同步版几乎一样：

@async_retry(RetryConfig(max_retries=3, base_delay=0.5))
async def fetch_async(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            response.raise_for_status()
            return await response.json()

关键区别：await asyncio.sleep(delay) 不会阻塞事件循环，其他协程可以继续执行。而 time.sleep 会阻塞整个线程，在异步代码里不能出现。

4.5 统一同步与异步的装饰器

有没有办法写一个装饰器，同时支持同步和异步函数？我试过几种方案，最干净的是用 inspect.iscoroutinefunction 判断函数类型。

import inspect

def universal_retry(config: Optional[RetryConfig] = None):
    if config is None:
        config = RetryConfig()
    
    def decorator(func):
        if inspect.iscoroutinefunction(func):
            @functools.wraps(func)
            async def async_wrapper(*args, **kwargs):
                # 复用异步重试逻辑
                last_exception = None
                for attempt in range(config.max_retries + 1):
                    try:
                        return await func(*args, **kwargs)
                    except config.retryable_exceptions as e:
                        last_exception = e
                        if attempt < config.max_retries:
                            delay = min(
                                config.base_delay * (config.backoff_factor ** attempt),
                                config.max_delay
                            )
                            if config.jitter:
                                import random
                                delay = delay * (0.5 + random.random())
                            await asyncio.sleep(delay)
                raise last_exception
            return async_wrapper
        else:
            @functools.wraps(func)
            def sync_wrapper(*args, **kwargs):
                # 复用同步重试逻辑
                last_exception = None
                for attempt in range(config.max_retries + 1):
                    try:
                        return func(*args, **kwargs)
                    except config.retryable_exceptions as e:
                        last_exception = e
                        if attempt < config.max_retries:
                            delay = min(
                                config.base_delay * (config.backoff_factor ** attempt),
                                config.max_delay
                            )
                            if config.jitter:
                                import random
                                delay = delay * (0.5 + random.random())
                            time.sleep(delay)
                raise last_exception
            return sync_wrapper
    return decorator

这样，同一个 @universal_retry 装饰器，既能用在同步函数上，也能用在异步函数上。我曾经在一个混合架构的项目里用这个方案，减少了大量重复代码。

4.6 重试策略核心流程

下面这张图展示了重试装饰器的完整执行流程，从调用函数到最终成功或失败。

重试装饰器执行流程 重试装饰器执行流程 调用被装饰函数 执行 func(*args, **kwargs) 成功? 返回结果 可重试 异常? 抛出异常 重试次数 用完? 抛出异常 计算退避延迟 等待 delay 秒 说明 • 成功时直接返回结果 • 不可重试异常直接抛出 • 重试次数用完也抛出 • 否则计算延迟后重试

4.7 避坑指南

我曾经在异步重试里犯过一个低级错误：用 time.sleep 代替 asyncio.sleep。结果整个事件循环卡住了，其他协程全部超时。嗯，这个坑踩过一次就再也不会忘了。

还有几个常见问题：

  • 重试次数过多：默认 3 次就够了，别设成 100 次。我见过一个配置把重试次数设成 10，结果下游服务挂了 10 分钟，这边还在疯狂重试，把日志都刷爆了。
  • 退避时间太长：max_delay 一定要设上限。指数退避如果不设上限，几次重试后延迟可能变成几分钟甚至几小时。
  • 异常类型太宽泛：不要对所有异常都重试。比如 ValueError、TypeError 这种业务异常，重试多少次都没用。

重要提醒：重试不是万能的。如果下游服务已经过载，你的重试只会加重问题。这时候应该考虑熔断和限流，而不是一味重试。

4.8 小结

← 上一章 📖 返回目录 下一章 →