Python脚本如何重试协程失败请求

wen python案例 2026-06-27 247

本文目录导读：

Python脚本如何重试协程失败请求

使用 asyncio 实现简单的重试
使用装饰器模式
使用 tenacity 库（推荐）
完整的示例：带监控和统计
选择建议

在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())