You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
81 lines
2.4 KiB
81 lines
2.4 KiB
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
性能优化模块
|
|
"""
|
|
import asyncio
|
|
import time
|
|
from typing import Dict, Any
|
|
from functools import wraps
|
|
|
|
from logger import get_logger
|
|
|
|
logger = get_logger("performance")
|
|
|
|
|
|
def monitor_performance(func):
|
|
"""性能监控装饰器"""
|
|
@wraps(func)
|
|
async def async_wrapper(*args, **kwargs):
|
|
start_time = time.time()
|
|
try:
|
|
result = await func(*args, **kwargs)
|
|
execution_time = time.time() - start_time
|
|
logger.info(f"{func.__name__} 执行完成，耗时: {execution_time:.2f}秒")
|
|
return result
|
|
except Exception as e:
|
|
execution_time = time.time() - start_time
|
|
logger.error(f"{func.__name__} 执行失败，耗时: {execution_time:.2f}秒，错误: {e}")
|
|
raise
|
|
|
|
@wraps(func)
|
|
def sync_wrapper(*args, **kwargs):
|
|
start_time = time.time()
|
|
try:
|
|
result = func(*args, **kwargs)
|
|
execution_time = time.time() - start_time
|
|
logger.info(f"{func.__name__} 执行完成，耗时: {execution_time:.2f}秒")
|
|
return result
|
|
except Exception as e:
|
|
execution_time = time.time() - start_time
|
|
logger.error(f"{func.__name__} 执行失败，耗时: {execution_time:.2f}秒，错误: {e}")
|
|
raise
|
|
|
|
if asyncio.iscoroutinefunction(func):
|
|
return async_wrapper
|
|
else:
|
|
return sync_wrapper
|
|
|
|
|
|
class PerformanceMonitor:
|
|
"""性能监控器"""
|
|
|
|
def __init__(self):
|
|
self.metrics: Dict[str, Any] = {}
|
|
self.start_time = time.time()
|
|
|
|
def start_timer(self, name: str):
|
|
"""开始计时"""
|
|
self.metrics[name] = {"start": time.time()}
|
|
|
|
def end_timer(self, name: str):
|
|
"""结束计时"""
|
|
if name in self.metrics:
|
|
self.metrics[name]["end"] = time.time()
|
|
self.metrics[name]["duration"] = (
|
|
self.metrics[name]["end"] - self.metrics[name]["start"]
|
|
)
|
|
logger.info(f"{name} 耗时: {self.metrics[name]['duration']:.2f}秒")
|
|
|
|
def get_summary(self) -> Dict[str, Any]:
|
|
"""获取性能摘要"""
|
|
total_time = time.time() - self.start_time
|
|
return {
|
|
"total_time": total_time,
|
|
"metrics": self.metrics,
|
|
"summary": f"总运行时间: {total_time:.2f}秒"
|
|
}
|
|
|
|
|
|
# 全局性能监控器
|
|
perf_monitor = PerformanceMonitor()
|
|
|