56 lines
1.6 KiB
Python
56 lines
1.6 KiB
Python
"""简单的重试工具：指数退避 + 抖动。
|
||
课程设计场景不引入额外依赖（如 tenacity），保持轻量。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import random
|
||
import time
|
||
from typing import Callable, TypeVar, Tuple
|
||
|
||
T = TypeVar("T")
|
||
|
||
|
||
def retry(
|
||
*,
|
||
max_retries: int = 3,
|
||
initial_delay: float = 1.0,
|
||
backoff_factor: float = 2.0,
|
||
max_delay: float = 10.0,
|
||
jitter: float = 0.2,
|
||
retry_exceptions: Tuple[type[BaseException], ...] = (Exception,),
|
||
):
|
||
"""重试装饰器。
|
||
|
||
- 第一次失败后等待 initial_delay
|
||
- 之后按照 backoff_factor 指数增长
|
||
- 加一点 jitter 防止固定间隔
|
||
"""
|
||
|
||
def decorator(func: Callable[..., T]) -> Callable[..., T]:
|
||
def wrapper(*args, **kwargs) -> T:
|
||
delay = initial_delay
|
||
last_exc: BaseException | None = None
|
||
|
||
for attempt in range(max_retries + 1):
|
||
try:
|
||
return func(*args, **kwargs)
|
||
except retry_exceptions as exc: # noqa: PERF203
|
||
last_exc = exc
|
||
if attempt >= max_retries:
|
||
raise
|
||
|
||
# jitter：delay*(1±jitter)
|
||
factor = 1.0 + random.uniform(-jitter, jitter)
|
||
sleep_s = min(max_delay, max(0.0, delay * factor))
|
||
time.sleep(sleep_s)
|
||
delay = min(max_delay, delay * backoff_factor)
|
||
|
||
# 理论上不会走到这里
|
||
assert last_exc is not None
|
||
raise last_exc
|
||
|
||
return wrapper
|
||
|
||
return decorator
|
||
|