文章
tenacity 重试监控实战
监控记录重试过程：
tools.py
import traceback
import logging
from tenacity import RetryCallState
# ---------- 全局日志配置 ----------
logging.basicConfig(
level=logging.INFO, # 可改为 WARNING 只显示重要信息
format="%(asctime)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)
# ---------- RetryMonitor 工具类 ----------
class RetryMonitor:
"""Tenacity 重试监控器，使用 logging 输出"""
def __init__(self, logger: logging.Logger = None):
self.logger = logger or logging.getLogger(__name__)
@staticmethod
def _get_error_detail(exception):
error_type = type(exception).__name__
error_msg = str(exception)
tb_lines = traceback.format_exc().split("\n")
error_location = "未知位置"
for line in reversed(tb_lines):
if line.strip().startswith("File"):
error_location = line.strip()
break
return error_type, error_msg, error_location
def before_sleep_callback(self, retry_state: RetryCallState) -> None:
"""重试前打印详细信息（before_sleep 回调）"""
exception = retry_state.outcome.exception()
error_type, error_msg, error_location = self._get_error_detail(exception)
wait_seconds = retry_state.next_action.sleep if retry_state.next_action else 0
msg = (
f"🔄 准备第 {retry_state.attempt_number} 次重试\n"
f"{'=' * 60}\n"
f"💥 上次错误类型: {error_type}\n"
f"📝 上次错误信息: {error_msg}\n"
f"📍 错误发生位置: {error_location}\n"
f"⏳ 等待 {wait_seconds:.1f} 秒后重试...\n"
f"{'=' * 60}"
)
self.logger.warning(msg)
def after_attempt_callback(self, retry_state: RetryCallState) -> None:
"""每次失败后打印（after 回调）"""
exception = retry_state.outcome.exception()
error_type, error_msg, _ = self._get_error_detail(exception)
msg = f"❌ 第 {retry_state.attempt_number} 次执行失败 -> {error_type}: {error_msg}"
self.logger.warning(msg)
test.py
import time
import random
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type, RetryError
)
from relearn_tenacity.tools import RetryMonitor
# 创建监控器（可全局共享）
monitor = RetryMonitor()
class FileProcessor:
"""模拟文件处理器"""
def __init__(self):
self.processed_count = 0
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=2, min=3, max=15),
retry=retry_if_exception_type((
TimeoutError,
ConnectionError,
OSError,
IOError,
)),
before_sleep=monitor.before_sleep_callback,
after=monitor.after_attempt_callback
)
def process_file(self, filename: str) -> dict:
"""处理文件"""
print(f"📄 开始处理文件: {filename}")
time.sleep(1)
fail_probability = random.random()
if fail_probability < 0.4:
raise TimeoutError(f"处理超时：文件 {filename} 过大，处理超过60秒")
elif fail_probability < 0.7:
raise IOError(f"IO错误：无法读取文件 {filename}，磁盘繁忙")
elif fail_probability < 0.85:
raise ConnectionError(f"连接中断：云存储服务临时不可用")
self.processed_count += 1
return {
"filename": filename,
"status": "processed",
"size": random.randint(1000, 50000)
}
def run_file_demo():
print("\n" + "=" * 35)
print("场景2: 文件处理自动重试")
print("=" * 35)
file_processor = FileProcessor()
files = ["data_batch_1.txt", "data_batch_2.txt", "data_batch_3.txt"]
for filename in files:
try:
result = file_processor.process_file(filename)
print(f"✅ {filename} 处理成功: 大小={result['size']}字节")
except RetryError as e:
print(f"❌ {filename} 处理失败!")
original_error = e.last_attempt.exception()
error_type = type(original_error).__name__
print(f" 错误类型: {error_type}")
print(f" 错误信息: {original_error}")
print("-" * 50)
if __name__ == '__main__':
run_file_demo()
输出：
===================================
场景2: 文件处理自动重试
===================================
📄 开始处理文件: data_batch_1.txt
2026-07-23 12:11:08 - ❌ 第 1 次执行失败 -> OSError: IO错误：无法读取文件 data_batch_1.txt，磁盘繁忙
2026-07-23 12:11:08 - 🔄 准备第 1 次重试
============================================================
💥 上次错误类型: OSError
📝 上次错误信息: IO错误：无法读取文件 data_batch_1.txt，磁盘繁忙
📍 错误发生位置: 未知位置
⏳ 等待 3.0 秒后重试...
============================================================
📄 开始处理文件: data_batch_1.txt
✅ data_batch_1.txt 处理成功: 大小=46202字节
--------------------------------------------------
📄 开始处理文件: data_batch_2.txt
✅ data_batch_2.txt 处理成功: 大小=16225字节
--------------------------------------------------
📄 开始处理文件: data_batch_3.txt
2026-07-23 12:11:14 - ❌ 第 1 次执行失败 -> TimeoutError: 处理超时：文件 data_batch_3.txt 过大，处理超过60秒
2026-07-23 12:11:14 - 🔄 准备第 1 次重试
============================================================
💥 上次错误类型: TimeoutError
📝 上次错误信息: 处理超时：文件 data_batch_3.txt 过大，处理超过60秒
📍 错误发生位置: 未知位置
⏳ 等待 3.0 秒后重试...
============================================================
📄 开始处理文件: data_batch_3.txt
2026-07-23 12:11:18 - ❌ 第 2 次执行失败 -> ConnectionError: 连接中断：云存储服务临时不可用
2026-07-23 12:11:18 - 🔄 准备第 2 次重试
============================================================
💥 上次错误类型: ConnectionError
📝 上次错误信息: 连接中断：云存储服务临时不可用
📍 错误发生位置: 未知位置
⏳ 等待 4.0 秒后重试...
============================================================
📄 开始处理文件: data_batch_3.txt
✅ data_batch_3.txt 处理成功: 大小=25740字节
--------------------------------------------------