本文目录导读：
我来介绍几种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库（功能最全面）
- 复杂场景：自定义重试处理器
- 需要回调：使用高级重试处理器
- 异步场景：使用异步重试处理器
这些封装方式都考虑了：
- 指数退避策略
- 异常类型过滤
- 日志记录
- 灵活配置
选择哪种方式取决于你的具体需求场景。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
