Python脚本重试策略指数退避怎么算

本文目录导读：

1. 基本公式
2. 带抖动的指数退避（推荐）
3. 全抖动策略（Full Jitter）
4. 等量/递减抖动（Equal/Decreasing Jitter）
5. 完整的重试函数实现
6. 手动控制的重试循环
7. 延迟增长示例
8. 选择建议

指数退避（Exponential Backoff）是一种常用的重试策略，重试间隔时间呈指数级增长,以下是几种常见的计算方式：

基本公式

import time
import random
def basic_backoff(attempt, base_delay=1, max_delay=60):
    """
    attempt: 当前重试次数（从0开始）
    base_delay: 基础延迟（秒）
    max_delay: 最大延迟（秒）
    """
    delay = min(base_delay * (2 ** attempt), max_delay)
    return delay
# 使用示例
for attempt in range(5):
    delay = basic_backoff(attempt)
    print(f"第{attempt+1}次重试，等待{delay}秒")
    # time.sleep(delay)

带抖动的指数退避（推荐）

def jittered_backoff(attempt, base_delay=1, max_delay=60, jitter_factor=0.1):
    """
    添加随机抖动，避免多个客户端同时重试造成"惊群效应"（thundering herd）
    """
    delay = min(base_delay * (2 ** attempt), max_delay)
    # 添加±jitter_factor%的随机抖动
    jitter = delay * jitter_factor * (2 * random.random() - 1)
    return delay + jitter
# 使用示例
for attempt in range(5):
    delay = jittered_backoff(attempt)
    print(f"第{attempt+1}次重试，等待{delay:.2f}秒")
    # time.sleep(delay)

全抖动策略（Full Jitter）

def full_jitter_backoff(attempt, base_delay=1, max_delay=60, cap=120):
    """
    Full Jitter: 在[0, cap]范围内随机，cap随重试次数增加
    Google API推荐使用
    """
    cap = min(base_delay * (2 ** attempt), max_delay)
    delay = random.uniform(0, cap)
    return min(delay, max_delay)
# 使用示例
for attempt in range(5):
    delay = full_jitter_backoff(attempt)
    print(f"第{attempt+1}次重试，等待{delay:.2f}秒")
    # time.sleep(delay)

等量/递减抖动（Equal/Decreasing Jitter）

def equal_jitter_backoff(attempt, base_delay=1, max_delay=60):
    """
    等量抖动：将延迟分成两半，一半固定，一半随机
    """
    temp = min(base_delay * (2 ** attempt), max_delay)
    half = temp / 2
    delay = half + random.uniform(0, half)
    return delay
# 使用示例
for attempt in range(5):
    delay = equal_jitter_backoff(attempt)
    print(f"第{attempt+1}次重试，等待{delay:.2f}秒")
    # time.sleep(delay)

完整的重试函数实现

import time
import random
from functools import wraps
def retry_with_backoff(
    max_retries=3, 
    base_delay=1, 
    max_delay=60, 
    backoff_factor=2,
    jitter=True,
    exceptions=(Exception,)
):
    """
    带指数退避的重试装饰器
    Args:
        max_retries: 最大重试次数
        base_delay: 基础延迟（秒）
        max_delay: 最大延迟（秒）
        backoff_factor: 退避因子（默认2）
        jitter: 是否启用抖动
        exceptions: 需要重试的异常类型
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_retries:
                        # 计算延迟
                        delay = min(
                            base_delay * (backoff_factor ** attempt), 
                            max_delay
                        )
                        # 添加抖动（默认启用）
                        if jitter:
                            delay = delay * (0.5 + random.random())
                        print(f"第{attempt+1}次重试，等待{delay:.2f}秒...")
                        time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator
# 使用示例
@retry_with_backoff(max_retries=3, base_delay=1, jitter=True)
def unstable_api_call():
    """模拟不稳定的API调用"""
    value = random.random()
    if value < 0.7:  # 70%概率失败
        raise ConnectionError("网络错误")
    return "成功"
# 测试
print(unstable_api_call())

手动控制的重试循环

import time
import random
def retry_operation(operation, max_retries=3, base_delay=1, max_delay=60):
    """手动控制重试流程"""
    for attempt in range(max_retries + 1):
        try:
            result = operation()
            return result
        except Exception as e:
            if attempt == max_retries:
                raise  # 最后一次失败，抛出异常
            # 指数退避计算
            delay = min(base_delay * (2 ** attempt), max_delay)
            # 添加抖动
            jitter = delay * 0.1 * (2 * random.random() - 1)
            actual_delay = delay + jitter
            print(f"第{attempt+1}次重试，等待{actual_delay:.2f}秒...")
            time.sleep(actual_delay)
# 使用示例
def my_operation():
    if random.random() < 0.6:
        raise ValueError("操作失败")
    return "操作成功"
result = retry_operation(my_operation, max_retries=5)
print(f"最终结果: {result}")

延迟增长示例

选择建议

1. 简单场景：使用基本指数退避
2. 高并发场景：必须添加抖动，推荐 Full Jitter
3. API调用：建议使用带抖动的指数退避
4. 关键服务：设置合理的最大延迟和重试次数上限

指数退避的核心是：重试次数越多，等待时间越长，同时通过抖动避免所有客户端同时重试。