244 lines
8.0 KiB
Python
244 lines
8.0 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
智能重试策略
|
||
包装其他策略并提供智能重试机制
|
||
"""
|
||
|
||
import asyncio
|
||
import time
|
||
import logging
|
||
from typing import Optional, List
|
||
from functools import wraps
|
||
|
||
from .base import IDownloadStrategy, DownloadTask, DownloadResult, TaskStatus
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class RetryStrategy(IDownloadStrategy):
|
||
"""智能重试策略，包装其他策略并提供重试机制"""
|
||
|
||
def __init__(
|
||
self,
|
||
strategy: IDownloadStrategy,
|
||
max_retries: int = 3,
|
||
retry_delays: Optional[List[float]] = None,
|
||
exponential_backoff: bool = True
|
||
):
|
||
"""
|
||
初始化重试策略
|
||
|
||
Args:
|
||
strategy: 被包装的策略
|
||
max_retries: 最大重试次数
|
||
retry_delays: 自定义重试延迟列表
|
||
exponential_backoff: 是否使用指数退避
|
||
"""
|
||
self.strategy = strategy
|
||
self.max_retries = max_retries
|
||
self.retry_delays = retry_delays or [1, 2, 5, 10, 30]
|
||
self.exponential_backoff = exponential_backoff
|
||
self.retry_stats = {
|
||
'total_retries': 0,
|
||
'successful_retries': 0,
|
||
'failed_retries': 0
|
||
}
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
return f"Retry({self.strategy.name})"
|
||
|
||
def get_priority(self) -> int:
|
||
"""继承被包装策略的优先级"""
|
||
return self.strategy.get_priority()
|
||
|
||
async def can_handle(self, task: DownloadTask) -> bool:
|
||
"""判断是否可以处理任务"""
|
||
return await self.strategy.can_handle(task)
|
||
|
||
async def download(self, task: DownloadTask) -> DownloadResult:
|
||
"""执行下载任务，带重试机制"""
|
||
original_retry_count = task.retry_count
|
||
last_error = None
|
||
|
||
for attempt in range(self.max_retries):
|
||
try:
|
||
# 更新任务状态
|
||
if attempt > 0:
|
||
task.status = TaskStatus.RETRYING
|
||
logger.info(f"任务 {task.task_id} 第 {attempt + 1}/{self.max_retries} 次重试")
|
||
|
||
# 执行下载
|
||
result = await self.strategy.download(task)
|
||
|
||
if result.success:
|
||
if attempt > 0:
|
||
self.retry_stats['successful_retries'] += 1
|
||
logger.info(f"任务 {task.task_id} 重试成功 (第 {attempt + 1} 次)")
|
||
return result
|
||
|
||
# 下载失败，准备重试
|
||
last_error = result.error_message
|
||
|
||
# 检查是否应该重试
|
||
if not self._should_retry(result, attempt):
|
||
logger.warning(f"任务 {task.task_id} 不符合重试条件，停止重试")
|
||
return result
|
||
|
||
# 计算延迟时间
|
||
delay = self._calculate_delay(attempt)
|
||
logger.info(f"任务 {task.task_id} 将在 {delay} 秒后重试")
|
||
await asyncio.sleep(delay)
|
||
|
||
# 增加重试计数
|
||
task.retry_count += 1
|
||
self.retry_stats['total_retries'] += 1
|
||
|
||
except Exception as e:
|
||
last_error = str(e)
|
||
logger.error(f"任务 {task.task_id} 执行异常: {e}")
|
||
|
||
if attempt < self.max_retries - 1:
|
||
delay = self._calculate_delay(attempt)
|
||
logger.info(f"任务 {task.task_id} 将在 {delay} 秒后重试")
|
||
await asyncio.sleep(delay)
|
||
task.retry_count += 1
|
||
self.retry_stats['total_retries'] += 1
|
||
else:
|
||
self.retry_stats['failed_retries'] += 1
|
||
break
|
||
|
||
# 所有重试都失败
|
||
task.status = TaskStatus.FAILED
|
||
self.retry_stats['failed_retries'] += 1
|
||
|
||
return DownloadResult(
|
||
success=False,
|
||
task_id=task.task_id,
|
||
error_message=f"重试 {self.max_retries} 次后仍然失败: {last_error}",
|
||
retry_count=task.retry_count
|
||
)
|
||
|
||
def _should_retry(self, result: DownloadResult, attempt: int) -> bool:
|
||
"""判断是否应该重试"""
|
||
# 如果已经达到最大重试次数，不重试
|
||
if attempt >= self.max_retries - 1:
|
||
return False
|
||
|
||
# 如果没有错误消息，可能是未知错误，应该重试
|
||
if not result.error_message:
|
||
return True
|
||
|
||
# 检查是否是可重试的错误
|
||
retryable_errors = [
|
||
'timeout',
|
||
'connection',
|
||
'network',
|
||
'429', # Too Many Requests
|
||
'503', # Service Unavailable
|
||
'502', # Bad Gateway
|
||
'504', # Gateway Timeout
|
||
'空响应',
|
||
'返回空',
|
||
'empty response',
|
||
'temporary'
|
||
]
|
||
|
||
error_lower = result.error_message.lower()
|
||
for error in retryable_errors:
|
||
if error in error_lower:
|
||
return True
|
||
|
||
# 检查是否是不可重试的错误
|
||
non_retryable_errors = [
|
||
'404', # Not Found
|
||
'403', # Forbidden
|
||
'401', # Unauthorized
|
||
'invalid',
|
||
'not found',
|
||
'deleted',
|
||
'已删除',
|
||
'不存在'
|
||
]
|
||
|
||
for error in non_retryable_errors:
|
||
if error in error_lower:
|
||
return False
|
||
|
||
# 默认重试
|
||
return True
|
||
|
||
def _calculate_delay(self, attempt: int) -> float:
|
||
"""计算重试延迟时间"""
|
||
if self.exponential_backoff:
|
||
# 指数退避：2^attempt 秒，最大30秒
|
||
delay = min(2 ** attempt, 30)
|
||
else:
|
||
# 使用预定义的延迟列表
|
||
if attempt < len(self.retry_delays):
|
||
delay = self.retry_delays[attempt]
|
||
else:
|
||
delay = self.retry_delays[-1]
|
||
|
||
# 添加一些随机性以避免同时重试
|
||
import random
|
||
jitter = random.uniform(0, 0.3 * delay)
|
||
|
||
return delay + jitter
|
||
|
||
def get_stats(self) -> dict:
|
||
"""获取重试统计信息"""
|
||
return self.retry_stats.copy()
|
||
|
||
def reset_stats(self):
|
||
"""重置统计信息"""
|
||
self.retry_stats = {
|
||
'total_retries': 0,
|
||
'successful_retries': 0,
|
||
'failed_retries': 0
|
||
}
|
||
|
||
|
||
def with_retry(
|
||
max_retries: int = 3,
|
||
retry_delays: Optional[List[float]] = None,
|
||
exponential_backoff: bool = True
|
||
):
|
||
"""
|
||
装饰器：为异步函数添加重试机制
|
||
|
||
Usage:
|
||
@with_retry(max_retries=3)
|
||
async def download_file(url):
|
||
...
|
||
"""
|
||
def decorator(func):
|
||
@wraps(func)
|
||
async def wrapper(*args, **kwargs):
|
||
last_exception = None
|
||
delays = retry_delays or [1, 2, 5, 10, 30]
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
return await func(*args, **kwargs)
|
||
except Exception as e:
|
||
last_exception = e
|
||
|
||
if attempt < max_retries - 1:
|
||
if exponential_backoff:
|
||
delay = min(2 ** attempt, 30)
|
||
else:
|
||
delay = delays[attempt] if attempt < len(delays) else delays[-1]
|
||
|
||
logger.warning(f"函数 {func.__name__} 失败 (尝试 {attempt + 1}/{max_retries}): {e}")
|
||
logger.info(f"将在 {delay} 秒后重试")
|
||
await asyncio.sleep(delay)
|
||
else:
|
||
logger.error(f"函数 {func.__name__} 重试 {max_retries} 次后仍然失败")
|
||
|
||
raise last_exception
|
||
|
||
return wrapper
|
||
return decorator |