脚本如何指数退避异步重试间隔

本文目录导读：

1. 基础指数退避实现 (Python)
2. 更完整的版本 (带装饰器)
3. JavaScript 实现
4. 高级版本：条件重试和自定义回调
5. 关键特性说明

基础指数退避实现 (Python)

import asyncio
import random
async def exponential_backoff_retry(
    func, 
    max_retries=3, 
    base_delay=1, 
    max_delay=60
):
    """通用指数退避重试函数"""
    for attempt in range(max_retries + 1):
        try:
            return await func()
        except Exception as e:
            if attempt == max_retries:
                raise e
            # 计算退避时间: base * 2^attempt + 随机抖动
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)  # 10% 抖动
            actual_delay = delay + jitter
            print(f"第 {attempt + 1} 次重试，等待 {actual_delay:.2f} 秒...")
            await asyncio.sleep(actual_delay)
# 使用示例
async def main():
    async def api_call():
        # 模拟 API 调用
        if random.random() < 0.7:  # 70% 概率失败
            raise ConnectionError("网络错误")
        return "成功"
    try:
        result = await exponential_backoff_retry(api_call, max_retries=5)
        print(f"结果: {result}")
    except Exception as e:
        print(f"最终失败: {e}")
asyncio.run(main())

更完整的版本 (带装饰器)

import asyncio
import random
from functools import wraps
from typing import Callable, Any, Optional
import time
class RetryConfig:
    def __init__(
        self,
        max_retries: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        jitter: bool = True,
        exponential_base: float = 2.0
    ):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.jitter = jitter
        self.exponential_base = exponential_base
def async_retry(config: Optional[RetryConfig] = None):
    """装饰器：指数退避重试"""
    if config is None:
        config = RetryConfig()
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            last_exception = None
            for attempt in range(config.max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt == config.max_retries:
                        raise e
                    # 计算延迟时间
                    delay = min(
                        config.base_delay * (config.exponential_base ** attempt),
                        config.max_delay
                    )
                    # 添加抖动
                    if config.jitter:
                        delay = delay + random.uniform(0, delay * 0.1)
                    print(f"[重试 {attempt + 1}/{config.max_retries}] "
                          f"错误: {e.__class__.__name__}: {e}, "
                          f"等待 {delay:.2f} 秒")
                    await asyncio.sleep(delay)
            raise last_exception
        return wrapper
    return decorator
# 使用装饰器
@async_retry(RetryConfig(max_retries=5, base_delay=0.5))
async def make_api_request(url: str):
    """模拟 API 请求"""
    if random.random() < 0.6:  # 60% 概率失败
        raise TimeoutError("请求超时")
    return {"status": "ok", "data": "示例数据"}
async def main():
    result = await make_api_request("https://api.example.com")
    print(f"API 响应: {result}")
asyncio.run(main())

JavaScript 实现

class ExponentialBackoff {
    constructor(options = {}) {
        this.maxRetries = options.maxRetries || 3;
        this.baseDelay = options.baseDelay || 1000; // 毫秒
        this.maxDelay = options.maxDelay || 60000;
        this.jitter = options.jitter !== undefined ? options.jitter : true;
        this.retryCount = 0;
    }
    // 计算下一次重试的延迟时间
    getDelay() {
        let delay = Math.min(
            this.baseDelay * Math.pow(2, this.retryCount),
            this.maxDelay
        );
        if (this.jitter) {
            delay += Math.random() * delay * 0.1;
        }
        return delay;
    }
    // 异步重试函数
    async retry(asyncFn, shouldRetry = () => true) {
        while (this.retryCount <= this.maxRetries) {
            try {
                const result = await asyncFn();
                return result;
            } catch (error) {
                this.retryCount++;
                if (this.retryCount > this.maxRetries || !shouldRetry(error)) {
                    throw error;
                }
                const delay = this.getDelay();
                console.log(`重试 ${this.retryCount}/${this.maxRetries}, 等待 ${delay}ms`);
                await new Promise(resolve => setTimeout(resolve, delay));
            }
        }
    }
    reset() {
        this.retryCount = 0;
    }
}
// 使用示例
async function makeRequest() {
    const backoff = new ExponentialBackoff({
        maxRetries: 5,
        baseDelay: 500,
        jitter: true
    });
    const result = await backoff.retry(async () => {
        const response = await fetch('https://api.example.com/data');
        if (!response.ok) {
            throw new Error(`HTTP ${response.status}`);
        }
        return response.json();
    }, (error) => {
        // 只重试特定类型的错误
        return error.message.includes('timeout') || error.message.includes('500');
    });
    return result;
}

高级版本：条件重试和自定义回调

import asyncio
import random
from typing import Callable, Optional, List, Type
import logging
logger = logging.getLogger(__name__)
class RetryHandler:
    def __init__(
        self,
        max_retries: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        exponential_base: float = 2.0,
        retry_on_exceptions: Optional[List[Type[Exception]]] = None,
        on_retry_callback: Optional[Callable] = None
    ):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.exponential_base = exponential_base
        self.retry_on_exceptions = retry_on_exceptions or [Exception]
        self.on_retry_callback = on_retry_callback
    async def execute(self, func: Callable, *args, **kwargs):
        """执行带重试的函数"""
        last_exception = None
        for attempt in range(self.max_retries + 1):
            try:
                return await func(*args, **kwargs)
            except tuple(self.retry_on_exceptions) as e:
                last_exception = e
                if attempt == self.max_retries:
                    logger.error(f"重试 {self.max_retries} 次后仍然失败: {e}")
                    raise
                # 计算退避延迟
                delay = min(
                    self.base_delay * (self.exponential_base ** attempt),
                    self.max_delay
                )
                # 添加随机抖动 (避免惊群效应)
                delay += random.uniform(0, delay * 0.1)
                if self.on_retry_callback:
                    await self.on_retry_callback(attempt + 1, delay, e)
                logger.info(f"第 {attempt + 1} 次重试，延迟 {delay:.2f} 秒")
                await asyncio.sleep(delay)
    async def execute_with_backoff(self, func: Callable, *args, **kwargs):
        """执行带有指数退避的重试"""
        return await self.execute(func, *args, **kwargs)
# 使用示例
async def make_network_request(url: str, retry_count: int = 0):
    """模拟网络请求"""
    if retry_count < 3:  # 模拟前几次失败
        raise ConnectionError("网络连接失败")
    return {"status": "success", "data": "response"}
class NetworkWrapper:
    def __init__(self):
        self.retry_handler = RetryHandler(
            max_retries=5,
            base_delay=0.5,
            max_delay=30,
            retry_on_exceptions=[ConnectionError, TimeoutError],
            on_retry_callback=self.on_retry
        )
    async def on_retry(self, attempt: int, delay: float, error: Exception):
        """重试回调函数"""
        print(f"计划重试 #{attempt}: 等待 {delay:.2f}s, 错误: {error}")
    async def fetch_data(self, url: str):
        """获取数据"""
        return await self.retry_handler.execute(make_network_request, url)
async def main():
    wrapper = NetworkWrapper()
    try:
        result = await wrapper.fetch_data("https://api.example.com")
        print(f"最终结果: {result}")
    except Exception as e:
        print(f"请求失败: {e}")
asyncio.run(main())

关键特性说明

选择哪种实现取决于你的具体需求：

• 简单场景：使用基础版本
• 需要装饰器：使用带装饰器的版本
• Web 开发：使用 JavaScript 版本
• 企业级应用：使用高级版本，支持更多自定义