告别Retrying！Tenacity的5个超实用功能让你的Python代码更健壮

如果你曾经在Python项目中处理过网络请求、数据库连接或者任何可能临时失败的操作，那么你一定对重试逻辑不陌生。几年前，很多开发者会首选retrying库来简化这部分工作，但自从它停止维护后，我们不得不寻找更现代、更强大的替代品。Tenacity就是这样一个库——它不仅仅是retrying的一个分叉，更是一次全面的进化。我在几个大型微服务项目中深度使用了Tenacity，发现它那些看似简单的功能组合起来，能解决很多实际开发中的棘手问题。今天我不打算重复那些基础教程，而是想分享五个真正能提升代码健壮性的高级功能，这些功能让我在构建高可用系统时省了不少心。

1. 复合等待策略：告别简单的线性等待

大多数重试库提供的等待策略都比较基础，比如固定间隔或者简单的指数退避。但在真实的生产环境中，网络抖动、服务负载、资源竞争等因素交织在一起，简单的策略往往效果不佳。Tenacity最让我欣赏的一点就是它提供了极其灵活的等待策略组合能力。

1.1 为什么需要复合策略？

想象一下这样的场景：你的服务需要调用一个第三方支付接口。这个接口在高峰期可能会因为负载过高而响应缓慢，但偶尔也会因为网络问题出现瞬时失败。如果使用固定间隔重试，你可能会在服务已经恢复后仍然等待过长时间；如果使用纯随机间隔，又可能在高负载时重试过于频繁，加重对方服务器负担。

Tenacity允许你将多种等待策略组合使用。比如，你可以先使用指数退避来应对可能的持续性问题，再加上一个随机抖动来避免多个客户端同时重试造成的"惊群效应"。

from tenacity import retry, wait_exponential, wait_random

@retry(
    wait=wait_exponential(multiplier=1, min=4, max=60) + wait_random(0, 2)
)
def call_payment_api(transaction_id, amount):
    """
    调用支付接口，使用复合等待策略
    - 指数退避：4, 8, 16, 32, 60, 60... 秒
    - 随机抖动：每次加上0-2秒的随机延迟
    """
    # 实际的API调用代码
    response = requests.post(
        "https://api.payment.com/charge",
        json={"id": transaction_id, "amount": amount},
        timeout=10
    )
    response.raise_for_status()
    return response.json()

这个组合策略的实际效果是：第一次重试等待4-6秒，第二次等待8-10秒，第三次等待16-18秒，以此类推。指数部分处理可能持续的服务降级，随机部分分散重试时间点。

1.2 更复杂的策略组合

在一些对延迟敏感但对成功率要求极高的场景中，你可能需要更精细的控制。比如在实时交易系统中，前几次重试应该尽快执行，但如果持续失败，则需要逐渐延长等待时间。

from tenacity import retry, wait_fixed, wait_random, wait_exponential
import random

@retry(
    wait=(
        wait_fixed(1) |  # 第一次快速重试
        wait_fixed(3) |  # 第二次稍慢
        wait_exponential(multiplier=2, min=5, max=30) + wait_random(0, 1)
    )
)
def execute_critical_transaction(order_data):
    """
    执行关键交易操作
    策略：快速尝试2次，然后进入指数退避+随机抖动
    """
    # 这里简化了实际的交易逻辑
    if random.random() < 0.3:  # 模拟30%的失败率
        raise ConnectionError("交易服务暂时不可用")
    return {"status": "success", "order_id": "12345"}

注意：|操作符在这里表示"或"的关系，Tenacity会按顺序尝试这些策略，直到其中一个满足条件。这种语法可能一开始看起来有点奇怪，但用习惯了会发现它非常直观。

我发现在实际项目中，最有效的策略往往是混合型的。下面这个表格总结了几种常见场景的推荐策略组合：

场景类型	推荐策略组合	核心考虑	典型等待时间序列
API调用	指数退避 + 随机抖动	避免客户端同步，应对临时过载	2-4s, 4-6s, 8-10s, 16-18s
数据库操作	固定间隔 + 条件退避	连接池问题，锁竞争	1s, 1s, 2s, 3s, 5s
文件/资源操作	固定间隔为主	资源释放需要时间	0.5s, 0.5s, 0.5s
分布式协调	随机指数退避	避免多个节点同时重试	随机1-3s, 随机2-6s, 随机4-12s

2. 运行时动态参数修改：让重试策略活起来

很多重试库的配置都是静态的——在装饰器里定义好，运行时就不能改了。但真实世界的系统是动态变化的：白天和夜晚的流量模式不同，工作日和周末的服务负载不同，甚至不同用户群体的请求特征也不同。Tenacity的.retry_with()方法让重试策略能够在运行时动态调整，这个功能在我处理多租户系统时特别有用。

2.1 基础动态调整

假设你有一个服务，需要根据当前系统负载自动调整重试策略。在低负载时，可以快速重试；在高负载时，应该延长重试间隔，避免雪崩效应。

from tenacity import retry, wait_fixed, stop_after_attempt
import psutil

# 基础的重试配置
@retry(
    wait=wait_fixed(1),
    stop=stop_after_attempt(3)
)
def process_user_request(user_id, data):
    """处理用户请求的基础函数"""
    # 这里是实际的处理逻辑
    return {"processed": True, "user_id": user_id}

# 根据系统负载动态调整重试策略
def get_adaptive_retry_function():
    """
    根据当前系统CPU使用率返回适配的重试函数
    """
    cpu_percent = psutil.cpu_percent(interval=0.1)
    
    if cpu_percent < 30:
        # 低负载：快速重试，最多5次
        return process_user_request.retry_with(
            wait=wait_fixed(0.5),
            stop=stop_after_attempt(5)
        )
    elif cpu_percent < 70:
        # 中等负载：标准策略
        return process_user_request.retry_with(
            wait=wait_fixed(1),
            stop=stop_after_attempt(3)
        )
    else:
        # 高负载：保守策略，减少重试频率
        return process_user_request.retry_with(
            wait=wait_fixed(3),
            stop=stop_after_attempt(2)
        )

# 使用动态调整后的函数
adaptive_function = get_adaptive_retry_function()
result = adaptive_function(user_id="123", data={"action": "update"})

2.2 基于业务规则的动态策略

更有趣的应用是根据业务规则动态调整策略。比如在电商系统中，处理VIP用户的订单时应该使用更积极的重试策略，而对普通用户则使用标准策略。

from tenacity import retry, wait_exponential, stop_after_delay
from datetime import datetime

class OrderProcessor:
    def __init__(self):
        # 基础装饰器配置
        self.base_retry = retry(
            wait=wait_exponential(multiplier=1, min=2, max=30),
            stop=stop_after_delay(60)  # 最多重试60秒
        )
    
    def _process_order_logic(self, order_data):
        """实际处理订单的逻辑"""
        # 模拟可能失败的操作
        if random.random() < 0.2:
            raise Exception("库存服务暂时不可用")
        return {"status": "processed", "order_id": order_data["id"]}
    
    def process_order(self, order_data):
        """根据订单类型动态调整重试策略"""
        user_tier = order_data.get("user_tier", "standard")
        order_amount = order_data.get("amount", 0)
        
        # 获取基础的重试函数
        processor = self.base_retry(self._process_order_logic)
        
        # 根据用户等级和订单金额调整策略
        if user_tier == "vip" or order_amount > 1000:
            # VIP用户或大额订单：更积极的重试
            processor = processor.retry_with(
                wait=wait_exponential(multiplier=0.5, min=1, max=15),
                stop=stop_after_delay(120)  # 重试时间延长到120秒
            )
        elif user_tier == "standard" and order_amount < 100:
            # 普通用户小额订单：更保守的策略
            processor = processor.retry_with(
                wait=wait_exponential(multiplier=2, min=5, max=60),
                stop=stop_after_delay(30)  # 只重试30秒
            )
        
        # 执行处理
        return processor(order_data)

# 使用示例
processor = OrderProcessor()

# VIP用户大额订单
vip_order = {
    "id": "order_001",
    "user_tier": "vip",
    "amount": 1500,
    "items": [...]
}
result1 = processor.process_order(vip_order)

# 普通用户小额订单
standard_order = {
    "id": "order_002",
    "user_tier": "standard",
    "amount": 50,
    "items": [...]
}
result2 = processor.process_order(standard_order)

这种基于业务规则的动态调整，让重试策略不再是简单的技术配置，而是成为了业务逻辑的一部分。我在实际项目中用这种方式处理过不同优先级的消息队列消费、不同重要性的数据同步任务等场景，效果很好。

3. 精细化的异常与返回值过滤

重试逻辑中最容易出错的地方之一就是"什么情况下应该重试"。如果重试了不该重试的异常，可能会掩盖真正的错误；如果该重试的没有重试，又会影响系统的健壮性。Tenacity提供了异常过滤和返回值过滤两种机制，让你可以精确控制重试的触发条件。

3.1 多层级异常过滤

在实际项目中，异常往往是有层级的。比如网络异常可能包含连接超时、连接拒绝、SSL错误等多种子类。Tenacity的retry_if_exception_type()支持异常类型的组合，让你可以构建精细化的异常过滤策略。

from tenacity import retry, retry_if_exception_type, stop_after_attempt
import requests
from requests.exceptions import (
    ConnectionError, Timeout, 
    ConnectTimeout, ReadTimeout, SSLError
)
from mysql.connector import Error as MySQLError
from mysql.connector.errors import (
    InterfaceError, DatabaseError,
    OperationalError, InternalError
)

# 定义可重试的异常类型
RETRYABLE_EXCEPTIONS = (
    # 网络相关异常
    ConnectionError | Timeout | ConnectTimeout | ReadTimeout | SSLError |
    # 数据库相关异常（部分）
    InterfaceError | OperationalError |
    # 自定义的业务异常
    TemporaryServiceError | RateLimitExceeded
)

@retry(
    retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
    stop=stop_after_attempt(5)
)
def fetch_data_with_retry(url, query_params=None):
    """
    获取数据，只对特定的可恢复异常进行重试
    """
    try:
        response = requests.get(
            url, 
            params=query_params,
            timeout=(3.05, 10)  # 连接超时3.05秒，读取超时10秒
        )
        response.raise_for_status()
        return response.json()
    
    except requests.exceptions.HTTPError as e:
        # HTTP错误码处理
        status_code = e.response.status_code
        
        if status_code in [429, 502, 503, 504]:
            # 这些状态码应该重试
            raise TemporaryServiceError(f"HTTP {status_code}: {str(e)}")
        else:
            # 其他HTTP错误不应该重试
            raise PermanentFailureError(f"HTTP {status_code}: {str(e)}")

提示：在设计异常过滤策略时，我建议创建一个专门的模块来管理可重试的异常类型。这样可以在整个项目中保持一致性，也便于后续调整。

3.2 基于返回值的条件重试

有些操作不会抛出异常，但返回的结果可能表示"需要重试"。比如调用一个API，它返回了{"status": "processing", "estimated_time": 30}，这意味着操作还在处理中，需要稍后重试获取最终结果。

from tenacity import retry, retry_if_result, stop_after_attempt, wait_fixed
from dataclasses import dataclass
from typing import Optional

@dataclass
class AsyncOperationResult:
    status: str  # "pending", "processing", "completed", "failed"
    result: Optional[dict] = None
    error_message: Optional[str] = None
    retry_after: Optional[int] = None  # 建议的重试等待时间

def should_retry_based_on_result(result: AsyncOperationResult) -> bool:
    """
    根据返回结果判断是否需要重试
    """
    # 这些状态表示操作还在进行中，应该重试
    if result.status in ["pending", "processing"]:
        return True
    
    # 特定的错误类型可以重试
    if result.status == "failed":
        error_msg = result.error_message or ""
        retryable_errors = [
            "timeout", "rate_limit", "temporary_unavailable",
            "resource_busy", "conflict"
        ]
        return any(err in error_msg.lower() for err in retryable_errors)
    
    return False

@retry(
    retry=retry_if_result(should_retry_based_on_result),
    stop=stop_after_attempt(10),
    wait=wait_fixed(2)
)
def poll_async_operation(operation_id: str) -> AsyncOperationResult:
    """
    轮询异步操作结果
    根据返回的状态决定是否继续重试
    """
    # 模拟API调用
    import random
    import time
    
    # 模拟网络延迟
    time.sleep(0.1)
    
    # 模拟不同的返回状态
    rand_val = random.random()
    
    if rand_val < 0.3:
        # 30%概率：还在处理中
        return AsyncOperationResult(
            status="processing",
            retry_after=2
        )
    elif rand_val < 0.6:
        # 30%概率：已完成
        return AsyncOperationResult(
            status="completed",
            result={"data": "operation_success"}
        )
    elif rand_val < 0.8:
        # 20%概率：临时失败
        return AsyncOperationResult(
            status="failed",
            error_message="Resource temporarily unavailable"
        )
    else:
        # 20%概率：永久失败
        return AsyncOperationResult(
            status="failed",
            error_message="Invalid operation ID"
        )

# 使用示例
try:
    final_result = poll_async_operation("op_123456")
    if final_result.status == "completed":
        print(f"操作成功: {final_result.result}")
    else:
        print(f"操作失败: {final_result.error_message}")
except Exception as e:
    print(f"重试耗尽: {e}")

这种基于返回值的重试逻辑特别适合处理异步操作、轮询任务、以及那些返回详细错误码而不是抛出异常的外部服务。

4. 企业级应用：数据库连接重试实战

数据库连接是分布式系统中最常见的故障点之一。网络闪断、连接池耗尽、数据库主从切换、临时过载等情况都会导致连接失败。一个健壮的系统必须能够优雅地处理这些临时故障。下面我分享一个在实际生产环境中使用的数据库连接重试方案。

4.1 智能化的数据库连接重试

简单的固定间隔重试对于数据库连接往往不够用。数据库故障的恢复时间可能从几毫秒（网络闪断）到几分钟（主从切换）不等。我们需要一个能够适应不同故障类型的智能重试策略。

from tenacity import (
    retry, retry_if_exception_type, 
    wait_exponential, wait_random,
    stop_after_attempt, stop_after_delay,
    before_sleep_log
)
import logging
import psycopg2
from psycopg2 import OperationalError, InterfaceError
import time

# 配置日志
logger = logging.getLogger(__name__)

class DatabaseConnectionManager:
    def __init__(self, host, port, database, user, password):
        self.connection_params = {
            "host": host,
            "port": port,
            "database": database,
            "user": user,
            "password": password,
            "connect_timeout": 5
        }
        self.conn = None
        
        # 定义可重试的数据库异常
        self.retryable_exceptions = (
            OperationalError |  # 操作错误，如连接失败
            InterfaceError     # 接口错误，如连接断开
        )
    
    @retry(
        retry=retry_if_exception_type(self.retryable_exceptions),
        wait=wait_exponential(multiplier=1, min=1, max=60) + wait_random(0, 2),
        stop=stop_after_attempt(10) | stop_after_delay(300),  # 最多10次或300秒
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True
    )
    def establish_connection(self):
        """
        建立数据库连接，带有智能重试策略
        """
        logger.info(f"尝试连接数据库 {self.connection_params['host']}:{self.connection_params['port']}")
        
        # 关闭现有连接（如果有）
        if self.conn and not self.conn.closed:
            try:
                self.conn.close()
            except:
                pass
        
        # 建立新连接
        self.conn = psycopg2.connect(**self.connection_params)
        
        # 设置连接参数
        self.conn.autocommit = False
        self.conn.set_session(readonly=False)
        
        logger.info("数据库连接建立成功")
        return self.conn
    
    @retry(
        retry=retry_if_exception_type(self.retryable_exceptions),
        wait=wait_exponential(multiplier=0.5, min=0.1, max=5),
        stop=stop_after_attempt(3),
        before_sleep=lambda retry_state: logger.debug(
            f"查询重试 #{retry_state.attempt_number}, "
            f"上次异常: {retry_state.outcome.exception() if retry_state.outcome else 'None'}"
        )
    )
    def execute_query(self, query, params=None):
        """
        执行查询，连接断开时自动重试
        """
        # 检查连接是否有效
        if self.conn is None or self.conn.closed:
            logger.warning("数据库连接已关闭，重新建立连接")
            self.establish_connection()
        
        try:
            with self.conn.cursor() as cursor:
                cursor.execute(query, params or ())
                
                if query.strip().upper().startswith("SELECT"):
                    return cursor.fetchall()
                else:
                    self.conn.commit()
                    return cursor.rowcount
                    
        except (OperationalError, InterfaceError) as e:
            logger.error(f"数据库操作失败: {e}")
            
            # 标记连接为无效
            if self.conn:
                try:
                    self.conn.close()
                except:
                    pass
                self.conn = None
            
            # 重新抛出异常，触发重试
            raise
    
    def get_connection_stats(self):
        """
        获取连接统计信息，用于监控和调试
        """
        if not self.conn:
            return {"status": "disconnected", "connection_count": 0}
        
        try:
            with self.conn.cursor() as cursor:
                cursor.execute("SELECT count(*) FROM pg_stat_activity WHERE pid <> pg_backend_pid()")
                other_connections = cursor.fetchone()[0]
                
                cursor.execute("SHOW max_connections")
                max_connections = cursor.fetchone()[0]
                
                return {
                    "status": "connected",
                    "other_connections": other_connections,
                    "max_connections": max_connections,
                    "connection_utilization": f"{(other_connections + 1) / int(max_connections) * 100:.1f}%"
                }
        except:
            return {"status": "error_getting_stats"}

# 使用示例
def main():
    # 初始化连接管理器
    db_manager = DatabaseConnectionManager(
        host="localhost",
        port=5432,
        database="myapp",
        user="app_user",
        password="secure_password"
    )
    
    # 建立连接（会自动重试）
    try:
        conn = db_manager.establish_connection()
        print("连接成功建立")
        
        # 执行查询（如果连接断开会自动重连）
        results = db_manager.execute_query(
            "SELECT * FROM users WHERE status = %s",
            ("active",)
        )
        print(f"查询到 {len(results)} 条记录")
        
        # 获取连接统计
        stats = db_manager.get_connection_stats()
        print(f"连接状态: {stats}")
        
    except Exception as e:
        logger.error(f"数据库操作最终失败: {e}")
        # 这里可以触发告警、降级逻辑等

4.2 连接池与重试的集成

在实际的企业级应用中，我们通常使用连接池来管理数据库连接。下面是一个将Tenacity与连接池集成的示例：

from tenacity import retry, retry_if_exception_type, wait_exponential, stop_after_attempt
from DBUtils.PooledDB import PooledDB
import pymysql
import threading
from contextlib import contextmanager

class ResilientConnectionPool:
    def __init__(self, host, user, password, database, 
                 pool_size=10, max_overflow=5):
        self.pool = PooledDB(
            creator=pymysql,
            host=host,
            user=user,
            password=password,
            database=database,
            maxconnections=pool_size,
            blocking=True,
            ping=1  # 每次连接时ping数据库检查连接
        )
        self._lock = threading.Lock()
        self._stats = {
            "total_connections": 0,
            "failed_connections": 0,
            "successful_retries": 0
        }
    
    @retry(
        retry=retry_if_exception_type((pymysql.OperationalError, pymysql.InterfaceError)),
        wait=wait_exponential(multiplier=1, min=0.5, max=10),
        stop=stop_after_attempt(3),
        reraise=True
    )
    def _get_connection_with_retry(self):
        """从连接池获取连接，失败时重试"""
        with self._lock:
            self._stats["total_connections"] += 1
        
        try:
            conn = self.pool.connection()
            # 测试连接是否真的可用
            with conn.cursor() as cursor:
                cursor.execute("SELECT 1")
                cursor.fetchone()
            
            return conn
            
        except (pymysql.OperationalError, pymysql.InterfaceError) as e:
            with self._lock:
                self._stats["failed_connections"] += 1
            raise
    
    @contextmanager
    def get_connection(self):
        """
        获取数据库连接的上下文管理器
        自动处理连接获取、重试和释放
        """
        conn = None
        try:
            conn = self._get_connection_with_retry()
            with self._lock:
                self._stats["successful_retries"] += 1
            yield conn
        finally:
            if conn:
                conn.close()
    
    def execute_with_retry(self, query, params=None, max_retries=2):
        """
        执行SQL语句，在语句级别也支持重试
        用于处理死锁、锁超时等临时错误
        """
        @retry(
            retry=retry_if_exception_type((
                pymysql.OperationalError,
                pymysql.InternalError,
                pymysql.DatabaseError
            )),
            wait=wait_exponential(multiplier=0.5, min=0.1, max=2),
            stop=stop_after_attempt(max_retries)
        )
        def _execute(connection, sql, args):
            with connection.cursor() as cursor:
                cursor.execute(sql, args or ())
                if sql.strip().upper().startswith("SELECT"):
                    return cursor.fetchall()
                else:
                    connection.commit()
                    return cursor.rowcount
        
        with self.get_connection() as conn:
            return _execute(conn, query, params)
    
    def get_pool_stats(self):
        """获取连接池统计信息"""
        with self._lock:
            return self._stats.copy()

# 使用示例
pool = ResilientConnectionPool(
    host="db.example.com",
    user="app_user",
    password="password",
    database="production_db",
    pool_size=20,
    max_overflow=10
)

# 执行查询（自动处理连接获取、重试、释放）
try:
    users = pool.execute_with_retry(
        "SELECT id, name, email FROM users WHERE active = %s",
        (1,)
    )
    print(f"Found {len(users)} active users")
    
    # 查看连接池统计
    stats = pool.get_pool_stats()
    print(f"Connection stats: {stats}")
    
except Exception as e:
    print(f"Database operation failed after retries: {e}")

这个方案的关键点在于：

1. 连接获取重试：在从连接池获取连接时进行重试
2. 语句执行重试：对具体的SQL语句执行进行重试，处理死锁等临时错误
3. 分层重试策略：连接级别和语句级别使用不同的重试参数
4. 统计监控：跟踪重试成功率，为容量规划提供数据支持

5. 高级监控与调试：让重试过程透明化

当重试逻辑变得复杂时，调试和监控就变得至关重要。你需要知道：重试发生了多少次？每次重试等待了多久？是什么原因触发的重试？Tenacity提供了丰富的钩子函数和回调机制，让你可以全面监控重试过程。

5.1 完整的重试生命周期监控

from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    before_sleep_log, after_log, before_log,
    retry_if_exception_type
)
import logging
import time
from dataclasses import dataclass
from typing import Dict, Any, Optional
import json

# 配置结构化日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class RetryMetrics:
    """重试指标收集"""
    operation_name: str
    start_time: float
    end_time: Optional[float] = None
    attempts: int = 0
    total_wait_time: float = 0.0
    exceptions: list = None
    success: bool = False
    
    def __post_init__(self):
        self.exceptions = []
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "operation": self.operation_name,
            "duration_seconds": self.end_time - self.start_time if self.end_time else None,
            "attempts": self.attempts,
            "total_wait_time": self.total_wait_time,
            "success": self.success,
            "exceptions": [str(e) for e in self.exceptions]
        }

class RetryMonitor:
    """重试监控器"""
    
    def __init__(self):
        self.metrics_store = {}
    
    def before_call(self, retry_state):
        """重试开始前的回调"""
        operation_id = id(retry_state.fn)
        if operation_id not in self.metrics_store:
            self.metrics_store[operation_id] = RetryMetrics(
                operation_name=retry_state.fn.__name__,
                start_time=time.time()
            )
        
        metrics = self.metrics_store[operation_id]
        metrics.attempts += 1
        
        logger.info(f"🔁 开始第 {retry_state.attempt_number} 次重试尝试 "
                   f"[函数: {retry_state.fn.__name__}]")
    
    def after_call(self, retry_state):
        """重试结束后的回调"""
        operation_id = id(retry_state.fn)
        if operation_id in self.metrics_store:
            metrics = self.metrics_store[operation_id]
            metrics.end_time = time.time()
            metrics.success = retry_state.outcome and retry_state.outcome.successful
            
            logger.info(f"{'✅' if metrics.success else '❌'} 重试结束 "
                       f"[函数: {retry_state.fn.__name__}, "
                       f"尝试次数: {metrics.attempts}, "
                       f"总耗时: {metrics.end_time - metrics.start_time:.2f}s]")
            
            # 发送到监控系统
            self._send_to_monitoring(metrics)
            
            # 清理
            del self.metrics_store[operation_id]
    
    def before_sleep(self, retry_state):
        """重试等待前的回调"""
        operation_id = id(retry_state.fn)
        if operation_id in self.metrics_store:
            metrics = self.metrics_store[operation_id]
            
            # 计算本次等待时间
            wait_time = retry_state.next_action.sleep
            metrics.total_wait_time += wait_time
            
            logger.info(f"⏸️  等待 {wait_time:.2f} 秒后重试 "
                       f"[函数: {retry_state.fn.__name__}, "
                       f"累计等待: {metrics.total_wait_time:.2f}s]")
    
    def on_exception(self, retry_state):
        """发生异常时的回调"""
        operation_id = id(retry_state.fn)
        if operation_id in self.metrics_store and retry_state.outcome:
            metrics = self.metrics_store[operation_id]
            if retry_state.outcome.failed:
                exception = retry_state.outcome.exception()
                metrics.exceptions.append(exception)
                
                logger.warning(f"⚠️  重试中捕获异常: {type(exception).__name__}: {str(exception)} "
                             f"[函数: {retry_state.fn.__name__}, 尝试: {retry_state.attempt_number}]")
    
    def _send_to_monitoring(self, metrics: RetryMetrics):
        """将指标发送到监控系统（示例）"""
        metrics_dict = metrics.to_dict()
        
        # 这里可以集成到Prometheus、Datadog等监控系统
        print(f"[监控] 重试指标: {json.dumps(metrics_dict, indent=2, ensure_ascii=False)}")
        
        # 示例：记录到文件或发送到监控API
        with open("retry_metrics.log", "a") as f:
            f.write(json.dumps(metrics_dict) + "\n")

# 创建监控器实例
monitor = RetryMonitor()

# 使用监控器的重试装饰器
def create_monitored_retry(**kwargs):
    """创建带有监控的重试装饰器"""
    return retry(
        **kwargs,
        before=monitor.before_call,
        after=monitor.after_call,
        before_sleep=monitor.before_sleep,
        retry_error_callback=monitor.on_exception
    )

# 示例：使用监控的重试
@create_monitored_retry(
    retry=retry_if_exception_type((ConnectionError, TimeoutError)),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    stop=stop_after_attempt(5)
)
def call_external_service(service_url: str, payload: dict):
    """
    调用外部服务，带有完整的重试监控
    """
    import random
    
    # 模拟外部服务调用
    if random.random() < 0.6:  # 60%失败率
        raise ConnectionError(f"无法连接到服务 {service_url}")
    
    return {"status": "success", "data": payload}

# 测试监控功能
if __name__ == "__main__":
    try:
        result = call_external_service(
            "https://api.example.com/data",
            {"query": "test"}
        )
        print(f"调用成功: {result}")
    except Exception as e:
        print(f"最终失败: {e}")

5.2 集成到现有监控体系

在实际生产环境中，你需要将重试监控集成到现有的监控体系中。下面是一个与Prometheus集成的示例：

from tenacity import retry, stop_after_attempt, wait_exponential
from prometheus_client import Counter, Histogram, Gauge
import time

# 定义Prometheus指标
RETRY_ATTEMPTS = Counter(
    'tenacity_retry_attempts_total',
    'Total number of retry attempts',
    ['function_name', 'exception_type']
)

RETRY_SUCCESS = Counter(
    'tenacity_retry_success_total',
    'Total number of successful retries',
    ['function_name']
)

RETRY_FAILURE = Counter(
    'tenacity_retry_failure_total',
    'Total number of failed retries',
    ['function_name', 'failure_reason']
)

RETRY_DURATION = Histogram(
    'tenacity_retry_duration_seconds',
    'Duration of retry operations',
    ['function_name'],
    buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60]
)

ACTIVE_RETRIES = Gauge(
    'tenacity_active_retries',
    'Number of currently active retry operations',
    ['function_name']
)

class PrometheusRetryMonitor:
    """Prometheus监控集成"""
    
    def __init__(self):
        self.active_operations = {}
    
    def before_call(self, retry_state):
        """重试开始"""
        func_name = retry_state.fn.__name__
        operation_id = id(retry_state)
        
        # 记录活跃重试
        ACTIVE_RETRIES.labels(function_name=func_name).inc()
        self.active_operations[operation_id] = {
            'function': func_name,
            'start_time': time.time()
        }
    
    def after_call(self, retry_state):
        """重试结束"""
        operation_id = id(retry_state)
        if operation_id in self.active_operations:
            func_name = self.active_operations[operation_id]['function']
            start_time = self.active_operations[operation_id]['start_time']
            
            # 减少活跃计数
            ACTIVE_RETRIES.labels(function_name=func_name).dec()
            
            # 记录持续时间
            duration = time.time() - start_time
            RETRY_DURATION.labels(function_name=func_name).observe(duration)
            
            # 记录成功/失败
            if retry_state.outcome and retry_state.outcome.successful:
                RETRY_SUCCESS.labels(function_name=func_name).inc()
            else:
                failure_reason = "unknown"
                if retry_state.outcome and retry_state.outcome.failed:
                    exc = retry_state.outcome.exception()
                    failure_reason = type(exc).__name__
                RETRY_FAILURE.labels(
                    function_name=func_name,
                    failure_reason=failure_reason
                ).inc()
            
            del self.active_operations[operation_id]
    
    def on_exception(self, retry_state):
        """发生异常"""
        if retry_state.outcome and retry_state.outcome.failed:
            func_name = retry_state.fn.__name__
            exc = retry_state.outcome.exception()
            
            RETRY_ATTEMPTS.labels(
                function_name=func_name,
                exception_type=type(exc).__name__
            ).inc()

# 创建监控器
prometheus_monitor = PrometheusRetryMonitor()

# 创建带有Prometheus监控的重试装饰器
def prometheus_retry(**kwargs):
    """创建带有Prometheus监控的重试装饰器"""
    return retry(
        **kwargs,
        before=prometheus_monitor.before_call,
        after=prometheus_monitor.after_call,
        retry_error_callback=prometheus_monitor.on_exception
    )

# 使用示例
@prometheus_retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10)
)
def process_order_payment(order_id: str, amount: float):
    """处理订单支付，带有Prometheus监控"""
    # 模拟支付处理
    import random
    if random.random() < 0.3:
        raise ConnectionError("支付网关连接失败")
    elif random.random() < 0.5:
        raise TimeoutError("支付处理超时")
    
    return {"status": "paid", "order_id": order_id}

# 在应用中集成
from prometheus_client import start_http_server

# 启动Prometheus指标服务器（通常在应用启动时调用）
def start_metrics_server(port=8000):
    """启动Prometheus指标服务器"""
    start_http_server(port)
    print(f"Metrics server started on port {port}")
    
    # 示例：暴露一些自定义指标
    from prometheus_client import generate_latest, REGISTRY
    from http.server import HTTPServer, BaseHTTPRequestHandler
    
    class MetricsHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            if self.path == '/metrics':
                self.send_response(200)
                self.send_header('Content-Type', 'text/plain')
                self.end_headers()
                self.wfile.write(generate_latest(REGISTRY))
            else:
                self.send_response(404)
                self.end_headers()
    
    server = HTTPServer(('', port), MetricsHandler)
    server.serve_forever()

这个监控方案提供了：

1. 实时指标：活跃重试数、重试持续时间
2. 历史统计：总重试次数、成功率、失败原因分布
3. 集成能力：可以轻松集成到Grafana、Datadog等监控平台
4. 告警基础：基于指标设置告警规则（如重试失败率过高）

在实际项目中，我通常会将这种监控与应用的业务指标结合起来，比如将支付重试失败率与订单失败率关联分析，从而更准确地定位问题根源。