本文目录导读：
我来介绍几种Python请求重试的封装方法,从简单到复杂：
基础重试装饰器
import time
import functools
from typing import Callable, Any, Type
def retry(
max_retries: int = 3,
delay: float = 1.0,
exceptions: tuple = (Exception,),
backoff: float = 2.0
) -> Callable:
"""
重试装饰器
Args:
max_retries: 最大重试次数
delay: 初始延迟时间(秒)
exceptions: 需要重试的异常类型
backoff: 延迟倍增因子
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
current_delay = delay
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_retries - 1:
print(f"重试第 {attempt + 1} 次，延迟 {current_delay}s")
time.sleep(current_delay)
current_delay *= backoff
raise last_exception
return wrapper
return decorator
# 使用示例
@retry(max_retries=3, delay=1.0, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url: str) -> dict:
import requests
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.json()
使用tenacity库（推荐）
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests
# 配置重试策略
retry_config = {
"stop": stop_after_attempt(3), # 最多重试3次
"wait": wait_exponential(multiplier=1, min=1, max=10), # 指数退避
"retry": retry_if_exception_type((ConnectionError, TimeoutError, requests.RequestException))
}
@retry(**retry_config)
def api_request(url: str, method: str = 'GET', **kwargs) -> requests.Response:
"""带重试的API请求"""
response = requests.request(method, url, **kwargs)
response.raise_for_status()
return response
# 使用示例
try:
result = api_request('https://api.example.com/data', timeout=5)
print("请求成功:", result.json())
except Exception as e:
print("所有重试都失败:", e)
完整的重试工具类
import time
import logging
from typing import Optional, Callable
from dataclasses import dataclass
from functools import wraps
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class RetryConfig:
"""重试配置"""
max_retries: int = 3
initial_delay: float = 1.0
max_delay: float = 60.0
backoff_factor: float = 2.0
jitter: bool = True # 添加随机抖动避免惊群效应
class RetryHandler:
"""重试处理器"""
def __init__(self, config: Optional[RetryConfig] = None):
self.config = config or RetryConfig()
def execute(self, func: Callable, *args, **kwargs):
"""执行带重试的函数"""
last_exception = None
current_delay = self.config.initial_delay
for attempt in range(self.config.max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < self.config.max_retries - 1:
# 计算延迟时间
delay = min(current_delay, self.config.max_delay)
# 添加随机抖动（±50%）
if self.config.jitter:
import random
delay = delay * (0.5 + random.random())
logger.warning(
f"请求失败 (尝试 {attempt + 1}/{self.config.max_retries}): {e}"
f"\n将在 {delay:.2f} 秒后重试..."
)
time.sleep(delay)
current_delay *= self.config.backoff_factor
# 所有重试都失败
logger.error(f"所有重试都失败，共尝试 {self.config.max_retries} 次")
raise last_exception
# 使用示例
retry_handler = RetryHandler(RetryConfig(max_retries=5))
# 方式1：直接使用
try:
result = retry_handler.execute(requests.get, 'https://api.example.com', timeout=5)
print("成功获取数据")
except Exception as e:
print(f"最终失败: {e}")
# 方式2：作为装饰器
def with_retry(handler: Optional[RetryHandler] = None):
"""重试装饰器工厂"""
handler = handler or RetryHandler()
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
return handler.execute(func, *args, **kwargs)
return wrapper
return decorator
@with_retry(RetryHandler(RetryConfig(max_retries=3)))
def fetch_user_data(user_id: int) -> dict:
"""获取用户数据"""
response = requests.get(f'https://api.example.com/users/{user_id}')
response.raise_for_status()
return response.json()
高级重试 - 带回调函数
from typing import Callable, Optional, List
from dataclasses import dataclass, field
@dataclass
class RetryCallback:
"""重试回调"""
on_retry: Optional[Callable] = None # 每次重试前回调
on_success: Optional[Callable] = None # 成功回调
on_failure: Optional[Callable] = None # 最终失败回调
on_attempt: Optional[Callable] = None # 每次尝试后回调
class AdvancedRetryHandler:
"""高级重试处理器"""
def __init__(
self,
config: RetryConfig,
callbacks: Optional[RetryCallback] = None,
should_retry: Optional[Callable[[Exception, int], bool]] = None
):
self.config = config
self.callbacks = callbacks or RetryCallback()
self.should_retry = should_retry or (lambda e, attempt: True)
def execute(self, func: Callable, *args, **kwargs):
"""执行带重试的函数"""
last_exception = None
current_delay = self.config.initial_delay
for attempt in range(self.config.max_retries):
try:
result = func(*args, **kwargs)
# 成功回调
if self.callbacks.on_success:
self.callbacks.on_success(result)
# 尝试后回调
if self.callbacks.on_attempt:
self.callbacks.on_attempt(attempt, True, result)
return result
except Exception as e:
last_exception = e
# 判断是否需要重试
if not self.should_retry(e, attempt):
raise
if attempt < self.config.max_retries - 1:
delay = min(current_delay, self.config.max_delay)
# 重试回调
if self.callbacks.on_retry:
self.callbacks.on_retry(e, attempt, delay)
# 尝试后回调
if self.callbacks.on_attempt:
self.callbacks.on_attempt(attempt, False, e)
time.sleep(delay)
current_delay *= self.config.backoff_factor
# 最终失败回调
if self.callbacks.on_failure:
self.callbacks.on_failure(last_exception, self.config.max_retries)
raise last_exception
# 使用示例
class APIClient:
"""API客户端"""
def __init__(self):
# 配置重试
config = RetryConfig(
max_retries=3,
initial_delay=1.0,
backoff_factor=2.0,
jitter=True
)
# 配置回调
callbacks = RetryCallback(
on_retry=lambda e, attempt, delay: logger.info(
f"正在重试 (第{attempt+1}次): {e}, 延迟{delay:.2f}s"
),
on_success=lambda result: logger.info("请求成功"),
on_failure=lambda e, retries: logger.error(
f"请求失败，已重试{retries}次: {e}"
),
on_attempt=lambda attempt, success, data: logger.debug(
f"尝试第{attempt+1}次: {'成功' if success else '失败'}"
)
)
# 判断是否需要重试
should_retry = lambda e, attempt: isinstance(e, (ConnectionError, TimeoutError))
self.retry_handler = AdvancedRetryHandler(config, callbacks, should_retry)
def get_data(self, url: str) -> dict:
"""获取数据"""
def _request():
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.json()
return self.retry_handler.execute(_request)
# 使用
client = APIClient()
try:
data = client.get_data('https://api.example.com/data')
print("数据获取成功:", data)
except Exception as e:
print("最终失败:", e)
异步重试实现
import asyncio
from typing import Callable, Any
import aiohttp
import logging
logger = logging.getLogger(__name__)
class AsyncRetryHandler:
"""异步重试处理器"""
def __init__(self, max_retries: int = 3, delay: float = 1.0):
self.max_retries = max_retries
self.delay = delay
async def execute(self, func: Callable, *args, **kwargs) -> Any:
"""异步执行带重试的函数"""
for attempt in range(self.max_retries):
try:
return await func(*args, **kwargs)
except Exception as e:
if attempt < self.max_retries - 1:
wait_time = self.delay * (2 ** attempt) # 指数退避
logger.warning(
f"异步请求失败 (尝试 {attempt + 1}/{self.max_retries}): {e}"
f"\n将在 {wait_time:.2f} 秒后重试..."
)
await asyncio.sleep(wait_time)
else:
logger.error(f"异步请求最终失败: {e}")
raise
# 使用示例
async def fetch_async_data(url: str) -> dict:
"""异步获取数据"""
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
response.raise_for_status()
return await response.json()
async def main():
retry_handler = AsyncRetryHandler(max_retries=3, delay=1.0)
try:
data = await retry_handler.execute(fetch_async_data, 'https://api.example.com')
print("异步获取成功:", data)
except Exception as e:
print("异步获取失败:", e)
# 运行异步代码
if __name__ == "__main__":
asyncio.run(main())
使用建议
- 简单场景：使用装饰器方式（第一种）
- 推荐使用：tenacity库（功能最全面）
- 复杂场景：自定义重试处理器
- 需要回调：使用高级重试处理器
- 异步场景：使用异步重试处理器
这些封装方式都考虑了：
- 指数退避策略
- 异常类型过滤
- 日志记录
- 灵活配置
选择哪种方式取决于你的具体需求场景。
