如何编写定位分析慢速代码片段脚本

wen 实用脚本 2026-06-24 243

本文目录导读：

如何编写定位分析慢速代码片段脚本

Python 代码性能分析脚本
行级代码分析工具
自动化慢代码检测脚本
完整示例：性能分析工作流
使用建议

我来为您介绍几种定位和分析慢速代码片段的方法和脚本编写方案。

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操作

这些脚本可以帮助您快速定位和分析代码中的性能瓶颈,根据具体需求选择合适的工具，持续优化代码性能。

本文地址： https://www.bmjup.com/post/7395.html