Tenacity 重试停止与等待策略

示例说明

重试策略的三个正交维度在此齐备：何时停（次数或总时长，先到为准）、等多久（指数增长加随机抖动封顶）、哪些错才重试（异常类型谓词）。装饰器把这些声明在函数签名上，调用方完全无感；before_sleep 钩子把每次重试前的情况写进日志，排障时能看到完整的尝试轨迹。

技术要点

• 停止条件用或运算组合双上限
• wait_random_exponential 附带抖动封顶
• reraise 让最终异常保持原始类型

代码解读

fetch_ledger 按随机数模拟三类结果：连接错误、限流错误、成功——前两类可重试；六个尝试上限与三十秒总时长先到即停。等待时长以 0.5 为乘数指数增长封顶十秒并加随机量。reraise 开关让耗尽后的异常是原始的 ConnectionError 而非包装错误，上层分支更干净。

使用建议

业务异常不要进可重试类型，白名单宁缺勿滥。总时长上限要纳入调用方的超时预算。抖动幅度按并发量调整防惊群。

代码示例

import random
from typing import Any

from tenacity import (
    RetryError,
    retry,
    stop_after_attempt,
    stop_after_delay,
    wait_random_exponential,
    before_sleep_log,
    retry_if_exception_type,
)
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class ThrottleError(Exception):
    pass


@retry(
    stop=(stop_after_attempt(6) | stop_after_delay(30)),
    wait=wait_random_exponential(multiplier=0.5, max=10),
    retry=retry_if_exception_type((ConnectionError, ThrottleError)),
    before_sleep=before_sleep_log(logger, logging.WARNING),
    reraise=True,
)
def fetch_ledger(page: int) -> list[dict[str, Any]]:
    roll = random.random()
    if roll < 0.35:
        raise ConnectionError('ledger node unreachable')
    if roll < 0.5:
        raise ThrottleError('quota exceeded for this window')
    return [{'page': page, 'rows': random.randint(1, 40)}]


def safe_fetch(page: int) -> list[dict[str, Any]]:
    try:
        return fetch_ledger(page)
    except RetryError:
        return []
    except (ConnectionError, ThrottleError):
        return []  # reraise=True 时最终以原始异常抛出


rows = safe_fetch(1)
print('ledger rows fetched:', len(rows))