Python 性能分析：工具与方法

Python 性能分析：工具与方法

1. 技术分析

1.1 性能分析概述

性能分析是定位代码瓶颈的关键：

性能分析层次
CPU分析: 定位CPU密集型操作
内存分析: 检测内存泄漏
IO分析: 发现IO瓶颈
线程分析: 排查并发问题

1.2 性能分析工具

1.3 性能指标

关键性能指标
执行时间: 完成任务所需时间
CPU利用率: CPU使用百分比
内存占用: 内存使用量
IO等待: 磁盘/网络等待时间

2. 核心功能实现

2.1 CPU 性能分析

import cProfile
import pstats

class CPUProfiler:
def __init__(self):
self.profiler = cProfile.Profile()

def profile(self, func, *args, **kwargs):
self.profiler.enable()
result = func(*args, **kwargs)
self.profiler.disable()

return result

def print_stats(self, sort_by='cumulative', top=10):
stats = pstats.Stats(self.profiler)
stats.sort_stats(sort_by)
stats.print_stats(top)

def save_stats(self, filename):
self.profiler.dump_stats(filename)

class LineProfilerWrapper:
def __init__(self):
try:
from line_profiler import LineProfiler
self.profiler = LineProfiler()
except ImportError:
raise ImportError("需要安装line_profiler: pip install line_profiler")

def profile_function(self, func):
self.profiler.add_function(func)

def run(self, cmd):
self.profiler.run(cmd)

def print_stats(self):
self.profiler.print_stats()

def profile_decorator(func):
def wrapper(*args, **kwargs):
profiler = cProfile.Profile()
profiler.enable()

try:
return func(*args, **kwargs)
finally:
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)

return wrapper

2.2 内存分析

class MemoryProfilerWrapper:
def __init__(self):
try:
from memory_profiler import memory_usage, profile
self.memory_usage = memory_usage
self.profile_decorator = profile
except ImportError:
raise ImportError("需要安装memory_profiler: pip install memory_profiler")

def measure_memory(self, func, *args, **kwargs):
mem_usage, result = self.memory_usage(
(func, args, kwargs),
interval=0.1,
retval=True
)

return result, max(mem_usage)

def profile(self, func):
return self.profile_decorator(func)

class MemoryAnalyzer:
def __init__(self):
self.allocations = []

def track_allocation(self, size, type_name):
self.allocations.append({
'size': size,
'type': type_name,
'timestamp': pd.Timestamp.now()
})

def get_top_consumers(self, n=10):
by_type = {}

for alloc in self.allocations:
by_type[alloc['type']] = by_type.get(alloc['type'], 0) + alloc['size']

return sorted(by_type.items(), key=lambda x: x[1], reverse=True)[:n]

2.3 性能监控

import time
import psutil

class PerformanceMonitor:
def __init__(self):
self.metrics = []

def collect(self):
process = psutil.Process()

metric = {
'timestamp': time.time(),
'cpu_percent': process.cpu_percent(),
'memory_percent': process.memory_percent(),
'memory_rss': process.memory_info().rss,
'num_threads': process.num_threads(),
'io_counters': process.io_counters()
}

self.metrics.append(metric)

return metric

def start_monitoring(self, interval=1):
import threading

def monitor():
while True:
self.collect()
time.sleep(interval)

thread = threading.Thread(target=monitor, daemon=True)
thread.start()

def get_summary(self):
if not self.metrics:
return {}

cpu_avg = sum(m['cpu_percent'] for m in self.metrics) / len(self.metrics)
mem_avg = sum(m['memory_percent'] for m in self.metrics) / len(self.metrics)

return {
'cpu_average': cpu_avg,
'memory_average': mem_avg,
'max_memory': max(m['memory_rss'] for m in self.metrics),
'total_samples': len(self.metrics)
}

class Timer:
def __init__(self):
self.start = None
self.end = None

def __enter__(self):
self.start = time.perf_counter()
return self

def __exit__(self, *args):
self.end = time.perf_counter()

@property
def elapsed(self):
if self.start is None:
return 0
end = self.end if self.end else time.perf_counter()
return end – self.start

3. 性能对比

3.1 分析工具对比

3.2 性能分析结果示例

3.3 内存分析结果示例

4. 最佳实践

4.1 性能分析流程

def analyze_performance(func, *args, **kwargs):
print("=== CPU分析 ===")
cpu_profiler = CPUProfiler()
result = cpu_profiler.profile(func, *args, **kwargs)
cpu_profiler.print_stats()

print("\\n=== 内存分析 ===")
mem_profiler = MemoryProfilerWrapper()
_, peak_mem = mem_profiler.measure_memory(func, *args, **kwargs)
print(f"峰值内存: {peak_mem:.2f} MB")

return result

class PerformanceAnalysisWorkflow:
def __init__(self, target_code):
self.target_code = target_code

def run(self):
print("1. 运行cProfile分析…")
self._run_cprofile()

print("\\n2. 定位热点函数…")
hot_functions = self._identify_hotspots()

print("\\n3. 行级分析热点函数…")
for func in hot_functions[:3]:
self._run_line_profiler(func)

print("\\n4. 内存分析…")
self._run_memory_profiler()

def _run_cprofile(self):
profiler = CPUProfiler()
profiler.profile(self.target_code)
profiler.print_stats()

def _identify_hotspots(self):
return []

def _run_line_profiler(self, func):
profiler = LineProfilerWrapper()
profiler.profile_function(func)
profiler.run(f"{func.__name__}()")
profiler.print_stats()

def _run_memory_profiler(self):
mem_profiler = MemoryProfilerWrapper()

@mem_profiler.profile
def wrapper():
self.target_code()

wrapper()

4.2 性能优化建议生成

class OptimizationSuggestionGenerator:
def __init__(self, profile_results):
self.profile_results = profile_results

def generate(self):
suggestions = []

for func, stats in self.profile_results.items():
if stats['cumulative_time'] > 1.0:
suggestions.append(f"优化 {func}: 累计耗时 {stats['cumulative_time']:.2f}s")

if stats['calls'] > 10000:
suggestions.append(f"{func} 调用次数过多 ({stats['calls']}次)，考虑缓存结果")

return suggestions

5. 总结

性能分析是优化的第一步：

对比数据如下：

• cProfile是最常用的性能分析工具
• line_profiler提供最精确的分析结果
• py-spy适合生产环境的性能监控
• 推荐先使用cProfile定位瓶颈，再使用line_profiler深入分析