本文目录导读：
我来分享几个Python异常容错封装的实用案例：
装饰器方式封装
基础重试装饰器
import time
import functools
from typing import Callable, Tuple, Type
def retry(max_retries: int = 3,
delay: float = 1,
exceptions: Tuple[Type[Exception], ...] = (Exception,)):
"""
异常重试装饰器
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_retries - 1:
raise
print(f"第{attempt+1}次失败: {e}")
time.sleep(delay)
return None
return wrapper
return decorator
# 使用示例
@retry(max_retries=3, delay=0.5, exceptions=(ValueError, ConnectionError))
def fetch_data(url: str) -> dict:
# 模拟异常
import random
if random.random() < 0.7:
raise ConnectionError("网络连接失败")
return {"data": "成功获取的数据"}
# 测试
result = fetch_data("http://example.com")
print(result)
兜底值装饰器
import functools
def fallback(default_value=None):
"""
异常时返回默认值
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
print(f"执行异常: {e}，返回默认值")
return default_value
return wrapper
return decorator
# 使用示例
@fallback(default_value="未知用户")
def get_username(user_id: int) -> str:
# 模拟可能异常的操作
if user_id <= 0:
raise ValueError("用户ID无效")
return f"用户{user_id}"
print(get_username(1)) # 正常
print(get_username(-1)) # 返回默认值
上下文管理器封装
超时容错上下文
import signal
import time
from contextlib import contextmanager
class TimeoutError(Exception):
pass
@contextmanager
def timeout(seconds: int):
"""
超时容错上下文管理器
"""
def handler(signum, frame):
raise TimeoutError("操作超时")
signal.signal(signal.SIGALRM, handler)
signal.alarm(seconds)
try:
yield
except TimeoutError as e:
print(f"超时容错: {e}")
# 这里可以进行补偿操作
except Exception as e:
print(f"其他异常容错: {e}")
finally:
signal.alarm(0) # 取消闹钟
# 使用示例
with timeout(seconds=2):
time.sleep(1) # 正常执行
with timeout(seconds=2):
time.sleep(3) # 触发超时容错
资源清理上下文
from contextlib import contextmanager
import sqlite3
@contextmanager
def safe_db_connection(db_path: str):
"""
数据库安全连接上下文管理器
"""
conn = None
try:
conn = sqlite3.connect(db_path)
yield conn
except sqlite3.Error as e:
print(f"数据库操作异常: {e}")
if conn:
conn.rollback() # 回滚事务
except Exception as e:
print(f"其他异常: {e}")
if conn:
conn.rollback()
finally:
if conn:
conn.close()
print("数据库连接已关闭")
# 使用示例
with safe_db_connection("test.db") as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM non_existent_table")
类封装方式
通用容错处理类
import functools
import logging
from typing import Callable, Any
class FaultTolerance:
"""
通用容错处理类
"""
def __init__(self,
max_retries: int = 3,
fallback_value: Any = None,
logger: logging.Logger = None):
self.max_retries = max_retries
self.fallback_value = fallback_value
self.logger = logger or logging.getLogger(__name__)
def retry(self, func: Callable, *args, **kwargs):
"""
重试机制
"""
last_exception = None
for attempt in range(self.max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
self.logger.warning(
f"第{attempt+1}次执行失败: {str(e)}"
)
if attempt < self.max_retries - 1:
import time
time.sleep(0.5 * (attempt + 1))
self.logger.error(
f"重试{self.max_retries}次后仍然失败"
)
return self.fallback_value
def safe_execute(self, func: Callable, *args, **kwargs):
"""
安全执行，捕获所有异常
"""
try:
return func(*args, **kwargs)
except KeyboardInterrupt:
raise # 不捕获键盘中断
except Exception as e:
self.logger.error(f"执行异常: {str(e)}")
return self.fallback_value
def __call__(self, func: Callable) -> Callable:
"""
作为装饰器使用
"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
return self.retry(func, *args, **kwargs)
return wrapper
# 配置日志
logging.basicConfig(level=logging.INFO)
# 使用示例
ft = FaultTolerance(max_retries=2, fallback_value=None)
@ft
def unstable_api_call(user_id: int) -> dict:
import random
if random.random() < 0.6:
raise ConnectionError("API不可用")
return {"user_id": user_id, "name": "Test"}
# 批量处理多个任务
tasks = [1, 2, 3, 4, 5]
results = []
for task in tasks:
result = ft.safe_execute(unstable_api_call, task)
results.append(result)
print(f"处理结果: {results}")
异步任务容错
import asyncio
import functools
from typing import Callable, Any
class AsyncFaultTolerance:
"""
异步任务容错处理
"""
def __init__(self,
max_retries: int = 3,
circuit_breaker_threshold: int = 5):
self.max_retries = max_retries
self.circuit_breaker_threshold = circuit_breaker_threshold
self.failure_count = 0
self.circuit_open = False
async def retry_async(self, func: Callable, *args, **kwargs):
"""
异步重试机制
"""
if self.circuit_open:
print("熔断器已打开，拒绝请求")
return None
for attempt in range(self.max_retries):
try:
if asyncio.iscoroutinefunction(func):
result = await func(*args, **kwargs)
else:
result = func(*args, **kwargs)
# 成功后重置故障计数
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
print(f"第{attempt+1}次失败: {e}")
# 检查熔断器
if self.failure_count >= self.circuit_breaker_threshold:
self.circuit_open = True
print("熔断器触发!")
break
await asyncio.sleep(0.5 * (attempt + 1))
return None
# 使用示例
async def unstable_async_api(url: str) -> dict:
import random
await asyncio.sleep(0.1)
if random.random() < 0.7:
raise ConnectionError(f"连接 {url} 失败")
return {"url": url, "status": "success"}
async def main():
ft = AsyncFaultTolerance(max_retries=3)
# 测试多个异步请求
urls = [f"http://api.example.com/{i}" for i in range(3)]
tasks = [ft.retry_async(unstable_async_api, url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"异步处理结果: {results}")
# 运行
asyncio.run(main())
配置文件驱动的容错
import json
from typing import Dict, Any
class ConfigDrivenFT:
"""
配置驱动的容错策略
"""
def __init__(self, config_path: str = None):
self.config = {
"retry": {
"enabled": True,
"max_retries": 3,
"delay": 1,
"backoff_factor": 2
},
"circuit_breaker": {
"enabled": True,
"failure_threshold": 5,
"recovery_timeout": 30
},
"fallback": {
"enabled": True,
"default_value": None
}
}
if config_path:
self.load_config(config_path)
def load_config(self, config_path: str):
"""加载配置文件"""
try:
with open(config_path, 'r') as f:
user_config = json.load(f)
self.config.update(user_config)
except FileNotFoundError:
print(f"配置文件 {config_path} 不存在，使用默认配置")
except json.JSONDecodeError:
print("配置文件格式错误，使用默认配置")
def execute_with_ft(self, func, *args, **kwargs):
"""根据配置执行容错策略"""
if self.config["retry"]["enabled"]:
return self._retry_strategy(func, *args, **kwargs)
else:
try:
return func(*args, **kwargs)
except Exception as e:
if self.config["fallback"]["enabled"]:
return self.config["fallback"]["default_value"]
raise
def _retry_strategy(self, func, *args, **kwargs):
"""重试策略实现"""
max_retries = self.config["retry"]["max_retries"]
delay = self.config["retry"]["delay"]
backoff = self.config["retry"]["backoff_factor"]
import time
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
if self.config["fallback"]["enabled"]:
return self.config["fallback"]["default_value"]
raise
wait_time = delay * (backoff ** attempt)
print(f"重试第{attempt+1}次，等待{wait_time}秒")
time.sleep(wait_time)
return self.config["fallback"]["default_value"]
# 配置文件示例 fault_tolerance.json
"""
{
"retry": {
"enabled": true,
"max_retries": 5,
"delay": 0.5,
"backoff_factor": 1.5
},
"fallback": {
"enabled": true,
"default_value": "fallback_data"
}
}
"""
# 使用示例
ft = ConfigDrivenFT("fault_tolerance.json")
def unstable_function(data: str) -> str:
import random
if random.random() < 0.6:
raise ValueError("模拟异常")
return f"处理: {data}"
result = ft.execute_with_ft(unstable_function, "测试数据")
print(f"容错处理结果: {result}")
最佳实践建议
- 分层容错：在不同层面（函数、模块、服务）实施合适的容错策略
- 日志记录：详细记录异常信息，便于调试和监控
- 适当重试：设置合理的重试次数和延迟策略
- 降级处理：提供备选方案或默认值
- 熔断机制：防止雪崩效应
- 监控告警：异常次数过多时及时告警
这些案例可以根据实际需求组合使用,构建健壮的容错系统。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
