本文目录导读：
在Python中重试协程失败的请求,有多种方法，这里介绍几种常用的方案：
使用 asyncio 实现简单的重试
import asyncio
import aiohttp
from typing import Callable, Any
async def retry_async(
func: Callable,
max_retries: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
exceptions: tuple = (Exception,)
) -> Any:
"""
通用异步重试装饰器
Args:
func: 要执行的异步函数
max_retries: 最大重试次数
delay: 初始延迟时间（秒）
backoff: 延迟增长倍数
exceptions: 需要重试的异常类型
"""
for attempt in range(max_retries):
try:
return await func()
except exceptions as e:
if attempt == max_retries - 1: # 最后一次尝试失败
raise
wait_time = delay * (backoff ** attempt)
print(f"尝试 {attempt + 1} 失败: {e}, {wait_time}秒后重试...")
await asyncio.sleep(wait_time)
# 使用示例
async def fetch_url(url: str) -> str:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
response.raise_for_status()
return await response.text()
async def main():
url = "https://api.example.com/data"
# 包装需要重试的函数
result = await retry_async(
lambda: fetch_url(url),
max_retries=5,
delay=1.0,
backoff=2.0,
exceptions=(aiohttp.ClientError, asyncio.TimeoutError)
)
print(result)
使用装饰器模式
import asyncio
from functools import wraps
from typing import Type, Tuple
def async_retry(
max_retries: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
exceptions: Tuple[Type[Exception], ...] = (Exception,)
):
"""
异步重试装饰器
"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt == max_retries - 1:
raise
wait_time = delay * (backoff ** attempt)
print(f"重试 {attempt + 1}/{max_retries}: {type(e).__name__}: {e}")
print(f"等待 {wait_time}秒...")
await asyncio.sleep(wait_time)
raise last_exception
return wrapper
return decorator
# 使用示例
@async_retry(max_retries=5, delay=0.5, backoff=1.5, exceptions=(aiohttp.ClientError,))
async def fetch_data(url: str):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
使用 tenacity 库（推荐）
tenacity 是最流行的重试库，支持异步：
import asyncio
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
# 配置重试策略
@retry(
stop=stop_after_attempt(5), # 最多重试5次
wait=wait_exponential(multiplier=1, min=1, max=10), # 指数退避：1, 2, 4, 8, 10秒
retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)),
reraise=True # 最终失败时抛出异常
)
async def fetch_with_retry(url: str):
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
response.raise_for_status()
return await response.json()
# 更高级的配置
from tenacity import before_sleep_log, after_log, stop_after_delay
import logging
logger = logging.getLogger(__name__)
@retry(
stop=stop_after_delay(30), # 最多重试30秒
wait=wait_exponential(multiplier=1, max=10),
retry=retry_if_exception_type((aiohttp.ClientError,)),
before_sleep=before_sleep_log(logger, logging.WARNING), # 重试前记录日志
after=after_log(logger, logging.INFO) # 重试后记录日志
)
async def advanced_fetch(url: str):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
完整的示例：带监控和统计
import asyncio
import aiohttp
import time
from typing import Optional
from dataclasses import dataclass
@dataclass
class RetryStats:
"""重试统计信息"""
total_attempts: int = 0
successful_attempts: int = 0
failed_attempts: int = 0
total_time: float = 0.0
class AsyncRetryClient:
"""异步重试客户端"""
def __init__(self,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 30.0,
timeout: float = 10.0):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.timeout = timeout
self.stats = RetryStats()
async def fetch(self, url: str, session: Optional[aiohttp.ClientSession] = None) -> str:
"""带重试的请求"""
start_time = time.time()
if session is None:
async with aiohttp.ClientSession() as session:
return await self._retry_request(session, url, start_time)
else:
return await self._retry_request(session, url, start_time)
async def _retry_request(self, session: aiohttp.ClientSession, url: str, start_time: float) -> str:
"""内部重试逻辑"""
for attempt in range(self.max_retries + 1):
self.stats.total_attempts += 1
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=self.timeout)) as response:
response.raise_for_status()
data = await response.text()
self.stats.successful_attempts += 1
self.stats.total_time = time.time() - start_time
return data
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
self.stats.failed_attempts += 1
if attempt == self.max_retries:
self.stats.total_time = time.time() - start_time
raise # 最终失败
# 计算等待时间（指数退避 + 随机抖动）
delay = min(self.base_delay * (2 ** attempt), self.max_delay)
# 添加随机抖动，避免所有客户端同时重试
import random
delay *= (1 + random.random() * 0.5)
print(f"请求失败: {type(e).__name__}: {e}")
print(f"等待 {delay:.2f}秒后重试 ({attempt + 1}/{self.max_retries})...")
await asyncio.sleep(delay)
def print_stats(self):
"""打印统计信息"""
print(f"\n--- 重试统计 ---")
print(f"总尝试次数: {self.stats.total_attempts}")
print(f"成功次数: {self.stats.successful_attempts}")
print(f"失败次数: {self.stats.failed_attempts}")
print(f"总耗时: {self.stats.total_time:.2f}秒")
# 使用示例
async def main():
client = AsyncRetryClient(max_retries=3, base_delay=1)
urls = [
"https://api.example.com/data1",
"https://api.example.com/data2",
]
async with aiohttp.ClientSession() as session:
tasks = [client.fetch(url, session) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
for url, result in zip(urls, results):
if isinstance(result, Exception):
print(f"{url}: 失败 - {result}")
else:
print(f"{url}: 成功 - {len(result)} 字符")
client.print_stats()
# 运行
if __name__ == "__main__":
asyncio.run(main())
选择建议
- 简单场景：使用自定义的重试函数或装饰器
- 生产环境：推荐使用
tenacity库，功能完善且稳定 - 需要监控：实现自定义的
AsyncRetryClient - 多个请求：使用
asyncio.gather并发执行
tenacity 安装：
pip install tenacity
