第四章：重试机制¶

为什么需要重试？¶

网络和服务可能临时故障：

网络抖动：短暂的网络问题
服务重启：服务正在重启
资源繁忙：临时资源不足
超时：请求处理超时

重试策略¶

固定间隔重试¶

import time

def retry_fixed(func, max_retries=3, delay=1):
    for i in range(max_retries):
        try:
            return func()
        except Exception as e:
            if i == max_retries - 1:
                raise e
            time.sleep(delay)

    return None

指数退避重试¶

def retry_exponential(func, max_retries=3, base_delay=1, max_delay=60):
    for i in range(max_retries):
        try:
            return func()
        except Exception as e:
            if i == max_retries - 1:
                raise e

            delay = min(base_delay * (2 ** i), max_delay)
            time.sleep(delay)

    return None

抖动重试¶

import random

def retry_jitter(func, max_retries=3, base_delay=1):
    for i in range(max_retries):
        try:
            return func()
        except Exception as e:
            if i == max_retries - 1:
                raise e

            # 添加随机抖动
            delay = base_delay * (2 ** i) + random.uniform(0, 1)
            time.sleep(delay)

    return None

使用装饰器¶

from functools import wraps
import time
import random

def retry(max_retries=3, base_delay=1, exceptions=(Exception,)):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if i == max_retries - 1:
                        raise e

                    delay = base_delay * (2 ** i) + random.uniform(0, 1)
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

# 使用
@retry(max_retries=3, base_delay=1, exceptions=(ConnectionError, TimeoutError))
def call_api():
    return requests.get("https://api.example.com")

使用 tenacity¶

pip install tenacity

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10)
)
def call_api():
    return requests.get("https://api.example.com")

# 指定异常
from tenacity import retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type(ConnectionError)
)
def call_api():
    pass

# 回调
from tenacity import after_log
import logging

@retry(
    stop=stop_after_attempt(3),
    after=after_log(logger, logging.WARNING)
)
def call_api():
    pass

重试最佳实践¶

1. 幂等性¶

# 确保操作幂等
def update_order(order_id, status):
    # 使用幂等键
    idempotency_key = f"order:{order_id}:{status}"

    if cache.get(idempotency_key):
        return  # 已处理

    db.update_order(order_id, status)
    cache.set(idempotency_key, 1, ttl=3600)

2. 限制重试次数¶

# 避免无限重试
@retry(stop=stop_after_attempt(3))
def call_api():
    pass

3. 合理选择异常¶

# 只重试可恢复的异常
@retry(retry=retry_if_exception_type((ConnectionError, TimeoutError)))
def call_api():
    pass

4. 记录日志¶

@retry(
    stop=stop_after_attempt(3),
    after=after_log(logger, logging.WARNING)
)
def call_api():
    pass

5. 配合熔断器¶

@retry(stop=stop_after_attempt(3))
@circuit_breaker(fail_max=5)
def call_api():
    pass

小结¶

本章学习了：

✅ 重试机制概念
✅ 重试策略
✅ 装饰器实现
✅ tenacity 库
✅ 最佳实践

恭喜完成高可用架构教程！ 🎉