python利用backoff实现异常自动重试详解

@backoff.on_exception(
    wait_gen=backoff.expo,  # 退避策略
    exception=(Exception,), # 需要捕获的异常类型（元组）
    max_tries=3,            # 最大重试次数（包括首次调用）
    max_time=30,            # 最大重试时间（秒）
    jitter=None,            # 随机抖动（防止并发请求同时重试）
    giveup=lambda e: False, # 可选：自定义放弃重试的条件
    logger=None,            # 日志记录器
    on_backoff=None,        # 重试时的回调函数
    on_giveup=None,         # 放弃重试时的回调函数
    factor=1,               # 退避策略的倍数因子
    **kwargs                # 其他参数
)

@backoff.on_exception(backoff.expo, RequestException, max_tries=5)
def fetch_data(url):
    return requests.get(url)

@backoff.on_exception(
    backoff.constant,
    KeyError,
    max_tries=3,
    interval=10  # 每次间隔 10 秒
)
def process_data(data):
    return data["key"]

@backoff.on_exception(
    backoff.linear,
    ConnectionError,
    max_tries=5,
    factor=2  # 每次间隔增加 2 秒
)
def connect_db():
    return connect()

import backoff
import requests
from requests.exceptions import RequestException

@backoff.on_exception(backoff.expo, RequestException, max_tries=5)
def get_api_data(url):
    response = requests.get(url)
    response.raise_for_status()  # 触发异常（如 4xx/5xx）
    return response.json()

try:
    data = get_api_data("https://api.example.com")
except RequestException as e:
    print(f"最终失败: {e}")

@backoff.on_exception(
    backoff.expo,
    (TimeoutError, ConnectionError),
    max_tries=3
)
def fetch_data():
    # 可能抛出 TimeoutError 或 ConnectionError
    pass

@backoff.on_exception(
    backoff.constant,
    KeyError,
    max_tries=3,
    interval=2  # 每次间隔 2 秒
)
def process_dict(data):
    return data["missing_key"]  # 触发 KeyError

@backoff.on_exception(
    backoff.expo,
    Exception,
    max_time=30  # 最大重试时间 30 秒
)
def unreliable_function():
    # 可能在 30 秒内多次重试
    pass

def giveup_on_404(exception):
    return getattr(exception, "status_code", 0) == 404

@backoff.on_exception(
    backoff.expo,
    RequestException,
    giveup=giveup_on_404
)
def get_data(url):
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

def log_backoff(details):
    print(f"重试 {details['tries']} 次，等待 {details['wait']} 秒")

def log_giveup(details):
    print(f"放弃重试: {details['value']}")

@backoff.on_exception(
    backoff.expo,
    Exception,
    on_backoff=log_backoff,
    on_giveup=log_giveup
)
def my_function():
    pass

@backoff.on_exception(backoff.expo, requests.exceptions.RequestException)
def get_http_response(url):
    return requests.get(url)

@backoff.on_exception(
    backoff.constant,
    sqlite3.OperationalError,
    max_tries=5,
    interval=1
)
def connect_db():
    return sqlite3.connect("my.db")

@backoff.on_exception(
    backoff.expo,
    FileNotFoundError,
    max_tries=3
)
def read_file(path):
    return open(path).read()

@backoff.on_exception(backoff.expo, Exception, max_tries=5)