本文目录导读：
我来为您介绍几种定位和分析慢速代码片段的方法和脚本编写方案。
Python 代码性能分析脚本
基础计时器装饰器
import time
import functools
import cProfile
import pstats
from typing import Any, Callable
def timer_decorator(func: Callable) -> Callable:
"""简单计时装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.perf_counter()
result = func(*args, **kwargs)
end_time = time.perf_counter()
elapsed = end_time - start_time
print(f"{func.__name__} 执行耗时: {elapsed:.4f} 秒")
return result
return wrapper
# 使用示例
@timer_decorator
def slow_function():
time.sleep(1)
return "完成"
slow_function()
高级性能分析器
import time
import functools
import tracemalloc
from collections import defaultdict
import threading
class PerformanceAnalyzer:
"""性能分析器"""
def __init__(self):
self.stats = defaultdict(lambda: {
'calls': 0,
'total_time': 0.0,
'min_time': float('inf'),
'max_time': 0.0
})
self._lock = threading.Lock()
def analyze(self, func: Callable) -> Callable:
"""分析函数性能"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 开始内存追踪
tracemalloc.start()
# 计时
start_time = time.perf_counter()
start_memory = tracemalloc.get_traced_memory()
try:
result = func(*args, **kwargs)
return result
finally:
end_time = time.perf_counter()
end_memory = tracemalloc.get_traced_memory()
# 计算统计
elapsed = end_time - start_time
memory_diff = end_memory[0] - start_memory[0]
# 更新统计信息（线程安全）
with self._lock:
stats = self.stats[func.__name__]
stats['calls'] += 1
stats['total_time'] += elapsed
stats['min_time'] = min(stats['min_time'], elapsed)
stats['max_time'] = max(stats['max_time'], elapsed)
stats['avg_time'] = stats['total_time'] / stats['calls']
stats['memory_used'] = memory_diff / 1024 # KB
tracemalloc.stop()
return wrapper
def report(self):
"""生成性能报告"""
print("\n" + "="*60)
print("性能分析报告")
print("="*60)
# 按总耗时排序
sorted_stats = sorted(
self.stats.items(),
key=lambda x: x[1]['total_time'],
reverse=True
)
for func_name, stats in sorted_stats:
print(f"\n函数: {func_name}")
print(f" 调用次数: {stats['calls']}")
print(f" 总耗时: {stats['total_time']:.4f}s")
print(f" 平均耗时: {stats['avg_time']:.4f}s")
print(f" 最小耗时: {stats['min_time']:.4f}s")
print(f" 最大耗时: {stats['max_time']:.4f}s")
if 'memory_used' in stats:
print(f" 内存使用: {stats['memory_used']:.2f} KB")
# 使用示例
analyzer = PerformanceAnalyzer()
@analyzer.analyze
def process_data(data_size=1000):
"""模拟数据处理"""
result = []
for i in range(data_size):
result.append(i ** 2)
time.sleep(0.001)
return result
# 测试
process_data(500)
process_data(1000)
process_data(2000)
analyzer.report()
行级代码分析工具
import line_profiler
import atexit
from io import StringIO
class LineProfiler:
"""行级性能分析器"""
def __init__(self):
self.profiler = line_profiler.LineProfiler()
atexit.register(self.print_stats)
def profile(self, func):
"""添加要分析的函数"""
self.profiler.add_function(func)
return func
def print_stats(self):
"""输出分析结果"""
if self.profiler.code_map:
s = StringIO()
self.profiler.print_stats(stream=s)
print(s.getvalue())
# 使用示例
profiler = LineProfiler()
@profiler.profile
def expensive_function():
"""需要优化的函数"""
total = 0
for i in range(10000):
total += sum([j * i for j in range(100)])
total *= 0.5
return total
@profiler.profile
def optimized_function():
"""优化后的函数"""
total = 0
for i in range(10000):
total += i * 5050 # 预计算 sum(0..99) = 4950
total *= 0.5
return total
expensive_function()
optimized_function()
自动化慢代码检测脚本
import time
import functools
import threading
import queue
from datetime import datetime
class SlowCodeDetector:
"""慢代码自动检测器"""
def __init__(self, threshold=0.5):
self.threshold = threshold # 耗时阈值（秒）
self.slow_functions = []
self._lock = threading.Lock()
def monitor(self, func):
"""监控函数执行时间"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.perf_counter()
try:
result = func(*args, **kwargs)
return result
finally:
elapsed = time.perf_counter() - start_time
if elapsed > self.threshold:
warning = {
'function': func.__name__,
'args': args,
'kwargs': kwargs,
'elapsed': elapsed,
'timestamp': datetime.now()
}
with self._lock:
self.slow_functions.append(warning)
print(f"⚠️ 警告: {func.__name__} 执行耗时 {elapsed:.4f}s，"
f"超过阈值 {self.threshold:.2f}s")
return wrapper
def generate_report(self):
"""生成慢代码报告"""
if not self.slow_functions:
print("✓ 未检测到慢代码")
return
print("\n" + "="*60)
print(f"慢代码检测报告 ({len(self.slow_functions)} 个问题)")
print("="*60)
# 按耗时排序
self.slow_functions.sort(key=lambda x: x['elapsed'], reverse=True)
for i, func_info in enumerate(self.slow_functions, 1):
print(f"\n{i}. 函数: {func_info['function']}")
print(f" 耗时: {func_info['elapsed']:.4f}s")
print(f" 时间: {func_info['timestamp'].strftime('%H:%M:%S')}")
print(f" 参数: {func_info['args'], func_info['kwargs']}")
# 使用示例
detector = SlowCodeDetector(threshold=0.3) # 设置300ms阈值
@detector.monitor
def fast_function():
time.sleep(0.1)
return "快函数"
@detector.monitor
def slow_function(n=100):
time.sleep(0.5)
result = [i ** 2 for i in range(n)]
return result
@detector.monitor
def very_slow_function(n=1000):
time.sleep(1.0)
result = [i ** 3 for i in range(n)]
return result
# 执行测试
fast_function()
slow_function(100)
very_slow_function(500)
detector.generate_report()
完整示例：性能分析工作流
import time
import random
import cProfile
import pstats
from io import StringIO
class PerformanceProfiler:
"""完整的性能分析工作流"""
def __init__(self, slow_threshold=0.1):
self.threshold = slow_threshold
self.results = {}
def profile_function(self, func, *args, **kwargs):
"""使用cProfile分析函数"""
# 创建Profile对象
profiler = cProfile.Profile()
# 开始分析
profiler.enable()
result = func(*args, **kwargs)
profiler.disable()
# 保存统计信息
s = StringIO()
ps = pstats.Stats(profiler, stream=s).sort_stats('cumulative')
ps.print_stats(20) # 打印前20个耗时操作
self.results[func.__name__] = {
'stats': s.getvalue(),
'result': result
}
return result
def analyze_detailed(self, code_string):
"""分析代码片段"""
import ast
# 解析代码
tree = ast.parse(code_string)
# 查找循环和函数调用
loops = [node for node in ast.walk(tree)
if isinstance(node, (ast.For, ast.While))]
print(f"代码复杂度分析:")
print(f" 循环数量: {len(loops)}")
# 估算复杂度
total_operations = 0
for loop in loops:
if isinstance(loop, ast.For):
# 估算迭代次数
total_operations += 100 # 假设平均100次
elif isinstance(loop, ast.While):
total_operations += 1000 # while循环更复杂
print(f" 估算操作次数: ~{total_operations}")
print(f" 建议优化: {'需要' if total_operations > 500 else '不需要'}")
def print_full_report(self):
"""打印完整报告"""
print("\n" + "="*60)
print("性能分析完整报告")
print("="*60)
for func_name, data in self.results.items():
print(f"\n函数: {func_name}")
print("-" * 40)
print(data['stats'][:500]) # 截断显示
# 使用示例
profiler = PerformanceProfiler()
# 定义需要分析的功能
def calculate_primes(limit=1000):
"""计算素数"""
primes = []
for num in range(2, limit):
is_prime = True
for i in range(2, int(num ** 0.5) + 1):
if num % i == 0:
is_prime = False
break
if is_prime:
primes.append(num)
return primes
def data_processing(data_size=1000):
"""数据处理"""
data = [random.random() for _ in range(data_size)]
# 多个操作
result1 = [x ** 2 for x in data]
result2 = [x ** 0.5 for x in data if x > 0.5]
result3 = sum(result1) / len(result1) if result1 else 0
return {'mean': result3, 'count': len(result2)}
# 执行分析
print("开始性能分析...\n")
profiler.profile_function(calculate_primes, 500)
profiler.profile_function(data_processing, 500)
profiler.print_full_report()
# 分析代码片段
code_example = """
def process(items):
result = []
for item in items:
for sub in item.children:
result.append(sub.value ** 2)
return result
"""
profiler.analyze_detailed(code_example)
使用建议
-
选择合适工具：
- 快速定位：使用
time模块计时 - 详细分析：使用
cProfile或line_profiler - 内存问题：使用
tracemalloc
- 快速定位：使用
-
阈值设置：
- 根据业务需求设定合理阈值
- Web应用：< 200ms
- 批处理：< 1s
-
持续监控：
- 集成到CI/CD流程
- 定期生成性能报告
-
优化策略：
- 使用缓存减少重复计算
- 优化算法复杂度
- 使用并行处理
- 减少不必要的I/O操作
这些脚本可以帮助您快速定位和分析代码中的性能瓶颈,根据具体需求选择合适的工具，持续优化代码性能。
