87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
import time
|
|
import psutil
|
|
import os
|
|
from typing import Dict, Any, Callable
|
|
from functools import wraps
|
|
|
|
class PerformanceMonitor:
|
|
"""性能监控器"""
|
|
|
|
def __init__(self):
|
|
self.metrics = {}
|
|
self.start_time = time.time()
|
|
|
|
def start_monitoring(self):
|
|
"""开始监控"""
|
|
self.start_time = time.time()
|
|
self.initial_memory = psutil.Process(os.getpid()).memory_info().rss
|
|
|
|
def stop_monitoring(self) -> Dict[str, Any]:
|
|
"""停止监控并返回性能指标"""
|
|
end_time = time.time()
|
|
final_memory = psutil.Process(os.getpid()).memory_info().rss
|
|
|
|
metrics = {
|
|
"execution_time": end_time - self.start_time,
|
|
"memory_used": final_memory - self.initial_memory,
|
|
"peak_memory": final_memory
|
|
}
|
|
|
|
return metrics
|
|
|
|
def monitor_function(self, func: Callable) -> Callable:
|
|
"""装饰器：监控函数性能"""
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
# 记录开始时间
|
|
start_time = time.time()
|
|
start_memory = psutil.Process(os.getpid()).memory_info().rss
|
|
|
|
try:
|
|
# 执行函数
|
|
result = func(*args, **kwargs)
|
|
return result
|
|
finally:
|
|
# 记录结束时间
|
|
end_time = time.time()
|
|
end_memory = psutil.Process(os.getpid()).memory_info().rss
|
|
|
|
# 记录指标
|
|
func_name = func.__name__
|
|
if func_name not in self.metrics:
|
|
self.metrics[func_name] = {
|
|
"call_count": 0,
|
|
"total_time": 0,
|
|
"total_memory": 0
|
|
}
|
|
|
|
self.metrics[func_name]["call_count"] += 1
|
|
self.metrics[func_name]["total_time"] += (end_time - start_time)
|
|
self.metrics[func_name]["total_memory"] += (end_memory - start_memory)
|
|
|
|
return wrapper
|
|
|
|
def get_metrics_report(self) -> str:
|
|
"""获取性能指标报告"""
|
|
if not self.metrics:
|
|
return "暂无性能数据"
|
|
|
|
report = "性能指标报告:\n"
|
|
report += "=" * 50 + "\n"
|
|
|
|
for func_name, metrics in self.metrics.items():
|
|
avg_time = metrics["total_time"] / metrics["call_count"]
|
|
avg_memory = metrics["total_memory"] / metrics["call_count"]
|
|
|
|
report += f"函数: {func_name}\n"
|
|
report += f" 调用次数: {metrics['call_count']}\n"
|
|
report += f" 总执行时间: {metrics['total_time']:.4f}s\n"
|
|
report += f" 平均执行时间: {avg_time:.4f}s\n"
|
|
report += f" 总内存使用: {metrics['total_memory'] / 1024 / 1024:.2f}MB\n"
|
|
report += f" 平均内存使用: {avg_memory / 1024 / 1024:.2f}MB\n"
|
|
report += "-" * 30 + "\n"
|
|
|
|
return report
|
|
|
|
# 创建全局性能监控器实例
|
|
perf_monitor = PerformanceMonitor() |