Python案例怎么实现异常重试？

本文目录导读：

1. 基础实现：使用循环
2. 使用装饰器模式
3. 使用第三方库：tenacity
4. 使用标准库的urllib3重试
5. 完整的重试工具类
6. 重试策略选择指南

Python中实现异常重试（Retry）有多种方法，从简单的循环到高级的重试库，下面我会介绍几种主要方案。

基础实现：使用循环

简单循环重试

import time
import random
def unreliable_function():
    """模拟一个可能失败的操作"""
    if random.random() < 0.7:  # 70%概率失败
        raise ConnectionError("网络连接失败")
    return "成功获取数据"
def retry_simple(func, max_retries=3, delay=1):
    """简单的重试函数"""
    for attempt in range(1, max_retries + 1):
        try:
            result = func()
            print(f"第{attempt}次尝试成功")
            return result
        except Exception as e:
            print(f"第{attempt}次尝试失败: {e}")
            if attempt == max_retries:
                raise  # 最后一次失败，抛出异常
            time.sleep(delay)
# 使用示例
try:
    result = retry_simple(unreliable_function, max_retries=5)
    print(f"最终结果: {result}")
except Exception as e:
    print(f"所有重试都失败: {e}")

带退避时间的重试

def retry_with_backoff(func, max_retries=3, base_delay=1, max_delay=60):
    """
    带指数退避的重试
    - base_delay: 基础延迟时间(秒)
    - max_delay: 最大延迟时间
    """
    for attempt in range(1, max_retries + 1):
        try:
            return func()
        except Exception as e:
            print(f"第{attempt}次尝试失败: {e}")
            if attempt == max_retries:
                raise
            # 指数退避 + 随机抖动
            delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
            jitter = random.uniform(0, delay * 0.1)
            total_delay = delay + jitter
            print(f"等待 {total_delay:.2f} 秒后重试...")
            time.sleep(total_delay)
# 使用示例
try:
    result = retry_with_backoff(unreliable_function, max_retries=5)
    print(f"最终结果: {result}")
except Exception as e:
    print(f"所有重试都失败: {e}")

使用装饰器模式

import functools
def retry_decorator(max_retries=3, delay=1, exceptions=(Exception,)):
    """
    重试装饰器
    :param max_retries: 最大重试次数
    :param delay: 重试间隔(秒)
    :param exceptions: 需要重试的异常类型元组
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    print(f"第{attempt}次尝试失败: {e}")
                    if attempt == max_retries:
                        raise
                    time.sleep(delay)
            return None
        return wrapper
    return decorator
# 使用装饰器
@retry_decorator(max_retries=5, delay=2, exceptions=(ConnectionError, TimeoutError))
def fetch_data():
    """从API获取数据"""
    if random.random() < 0.6:
        raise ConnectionError("API连接失败")
    return {"status": "success", "data": [1, 2, 3]}
# 调用
try:
    result = fetch_data()
    print(f"获取数据成功: {result}")
except Exception as e:
    print(f"最终失败: {e}")

使用第三方库：tenacity

tenacity 是最流行的Python重试库，功能强大且易于使用。

安装

pip install tenacity

基本使用

from tenacity import retry, stop_after_attempt, wait_fixed
@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_fixed(2)  # 每次等待2秒
)
def fetch_data():
    """尝试获取数据"""
    if random.random() < 0.6:
        raise ConnectionError("连接失败")
    return "数据获取成功"
# 调用
try:
    result = fetch_data()
    print(f"成功: {result}")
except Exception as e:
    print(f"失败: {e}")

高级用法

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests
@retry(
    stop=stop_after_attempt(5),  # 最多重试5次
    wait=wait_exponential(multiplier=1, min=2, max=30),  # 指数退避
    retry=retry_if_exception_type((ConnectionError, TimeoutError, requests.RequestException)),
    before_sleep=lambda retry_state: print(f"等待{retry_state.next_action.sleep}秒后重试..."),
    after=lambda retry_state: print(f"第{retry_state.attempt_number}次尝试, 结果: {retry_state.outcome}")
)
def call_api(url, timeout=5):
    """调用API接口"""
    response = requests.get(url, timeout=timeout)
    response.raise_for_status()
    return response.json()
# 使用
try:
    data = call_api("https://api.example.com/data")
    print(f"API调用成功: {data}")
except Exception as e:
    print(f"API调用最终失败: {e}")

使用标准库的urllib3重试

from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
import requests
# 配置重试策略
retry_strategy = Retry(
    total=3,  # 总重试次数
    backoff_factor=1,  # 退避因子
    status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的HTTP状态码
    allowed_methods=["GET", "POST"]  # 允许重试的方法
)
# 创建会话
session = requests.Session()
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
# 使用会话
try:
    response = session.get("https://api.example.com/data", timeout=10)
    response.raise_for_status()
    print(f"请求成功: {response.json()}")
except Exception as e:
    print(f"请求失败: {e}")

完整的重试工具类

import time
import random
import logging
from typing import Callable, Type, Tuple, Any
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RetryHandler:
    """重试处理器"""
    def __init__(
        self,
        max_retries: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        exponential_backoff: bool = True,
        retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,)
    ):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.exponential_backoff = exponential_backoff
        self.retryable_exceptions = retryable_exceptions
    def execute(self, func: Callable, *args, **kwargs) -> Any:
        """执行带重试的函数"""
        last_exception = None
        for attempt in range(1, self.max_retries + 1):
            try:
                return func(*args, **kwargs)
            except self.retryable_exceptions as e:
                last_exception = e
                logger.warning(f"第{attempt}次尝试失败: {e}")
                if attempt == self.max_retries:
                    logger.error(f"达到最大重试次数({self.max_retries})")
                    raise
                # 计算等待时间
                if self.exponential_backoff:
                    delay = min(
                        self.base_delay * (2 ** (attempt - 1)),
                        self.max_delay
                    )
                else:
                    delay = self.base_delay
                # 增加随机抖动
                delay *= 1 + random.uniform(-0.1, 0.1)
                logger.info(f"等待 {delay:.2f} 秒后重试...")
                time.sleep(delay)
    def __call__(self, func: Callable) -> Callable:
        """作为装饰器使用"""
        from functools import wraps
        @wraps(func)
        def wrapper(*args, **kwargs):
            return self.execute(func, *args, **kwargs)
        return wrapper
# 使用示例
retry_handler = RetryHandler(
    max_retries=5,
    base_delay=1,
    max_delay=30,
    retryable_exceptions=(ConnectionError, TimeoutError, ValueError)
)
# 方式1：直接调用
def risky_operation(value):
    if random.random() < 0.6:
        raise ConnectionError(f"操作失败: {value}")
    return f"成功处理: {value}"
try:
    result = retry_handler.execute(risky_operation, "test_data")
    print(f"结果: {result}")
except Exception as e:
    print(f"最终失败: {e}")
# 方式2：作为装饰器
@retry_handler
def another_risky_operation():
    if random.random() < 0.5:
        raise TimeoutError("超时")
    return "操作成功"
try:
    result = another_risky_operation()
    print(f"结果: {result}")
except Exception as e:
    print(f"最终失败: {e}")

重试策略选择指南

核心建议：

• 总是使用指数退避策略,避免雪崩效应
• 设置合理的最大重试次数和超时时间
• 明确哪些异常需要重试（区分可恢复和不可恢复异常）
• 记录重试日志便于问题排查
• 考虑添加熔断机制防止持续失败

根据实际需求选择最适合的重试方案即可。