性能分析与调优

性能分析是优化代码的第一步。Python提供了多种工具来帮助你找到性能瓶颈并进行优化。

timeit：快速计时

timeit适合测量小段代码的执行时间。

import timeit

# 基本用法
time_taken = timeit.timeit('sum(range(1000))', number=10000)
print(f"执行10000次sum(range(1000)): {time_taken:.4f}秒")

# 测量单次执行时间
single_time = timeit.timeit('sum(range(1000))', number=1)
print(f"单次执行时间: {single_time:.6f}秒")

# 使用setup
setup = """
import random
data = [random.randint(0, 100) for _ in range(1000)]
"""
time_taken = timeit.timeit('sorted(data)', setup=setup, number=1000)
print(f"排序1000个元素1000次: {time_taken:.4f}秒")

比较不同实现

import timeit

# 比较列表推导式和map
setup = "data = list(range(1000))"

list_comp_time = timeit.timeit(
    '[x ** 2 for x in data]', 
    setup=setup, 
    number=10000
)

map_time = timeit.timeit(
    'list(map(lambda x: x ** 2, data))', 
    setup=setup, 
    number=10000
)

print(f"列表推导式: {list_comp_time:.4f}秒")
print(f"map函数: {map_time:.4f}秒")
print(f"列表推导式快 {map_time/list_comp_time:.2f} 倍")

cProfile：函数级分析

cProfile提供函数调用的详细统计信息。

import cProfile
import pstats
from io import StringIO

def fibonacci_recursive(n):
    """递归斐波那契（低效）"""
    if n <= 1:
        return n
    return fibonacci_recursive(n - 1) + fibonacci_recursive(n - 2)

def fibonacci_iterative(n):
    """迭代斐波那契（高效）"""
    if n <= 1:
        return n
    a, b = 0, 1
    for _ in range(2, n + 1):
        a, b = b, a + b
    return b

def main():
    # 测试递归版本
    for i in range(30):
        fibonacci_recursive(i)
    
    # 测试迭代版本
    for i in range(30):
        fibonacci_iterative(i)

# 分析性能
profiler = cProfile.Profile()
profiler.enable()
main()
profiler.disable()

# 打印统计信息
print("性能分析结果:")
stats = pstats.Stats(profiler)
stats.strip_dirs()
stats.sort_stats('cumulative')
stats.print_stats(10)  # 显示前10个最耗时的函数

保存和分析结果

import cProfile
import pstats
import io

def cpu_intensive_function():
    """CPU密集型函数"""
    total = 0
    for i in range(1000000):
        total += i ** 2
    return total

# 分析并保存结果
profiler = cProfile.Profile()
profiler.enable()
result = cpu_intensive_function()
profiler.disable()

# 保存到文件
profiler.dump_stats('profile_output.prof')

# 从文件加载并分析
stats = pstats.Stats('profile_output.prof')
stats.strip_dirs()
stats.sort_stats('tottime')
stats.print_stats(20)

line_profiler：逐行分析

line_profiler可以分析每一行代码的执行时间。

# 安装: pip install line_profiler

# 创建要分析的文件 (example.py)
def process_data(data):
    """处理数据的函数"""
    result = []
    for item in data:
        if item % 2 == 0:
            result.append(item ** 2)
    return result

def main():
    data = list(range(10000))
    result = process_data(data)
    return sum(result)

if __name__ == "__main__":
    main()

使用line_profiler

# 在命令行中使用:
# kernprof -l -v example.py

# 或者在代码中使用
from line_profiler import LineProfiler

def process_data(data):
    result = []
    for item in data:
        if item % 2 == 0:
            result.append(item ** 2)
    return result

def main():
    data = list(range(10000))
    result = process_data(data)
    return sum(result)

# 创建LineProfiler实例
lp = LineProfiler()
lp.add_function(process_data)
lp.add_function(main)

# 包装要分析的函数
lp_wrapper = lp(main)

# 运行并打印结果
lp_wrapper()
lp.print_stats()

memory_profiler：内存分析

memory_profiler可以逐行分析内存使用情况。

# 安装: pip install memory_profiler

# 创建要分析的文件 (memory_example.py)
@profile  # 这个装饰器会在分析时自动添加
def process_large_data():
    """处理大量数据"""
    # 创建大列表
    data = [i for i in range(1000000)]
    
    # 处理数据
    processed = [x ** 2 for x in data]
    
    # 过滤数据
    filtered = [x for x in processed if x % 2 == 0]
    
    # 返回结果
    return sum(filtered)

if __name__ == "__main__":
    process_large_data()

使用memory_profiler

# 在命令行中使用:
# python -m memory_profiler memory_example.py

# 输出示例:
# Line #    Mem usage    Increment   Line Contents
# ================================================
#      3     38.5 MiB     38.5 MiB   def process_large_data():
#      4                         """处理大量数据"""
#      5     74.7 MiB     36.2 MiB       data = [i for i in range(1000000)]
#      6                        0.0 MiB       processed = [x ** 2 for x in data]
#      7                        0.0 MiB       filtered = [x for x in processed if x % 2 == 0]
#      8                        0.0 MiB       return sum(filtered)

py-spy：采样分析器

py-spy是一个采样分析器，可以在不修改代码的情况下分析性能。

# 安装: pip install py-spy

# 使用方法:
# py-spy top --pid <PID>
# py-spy record -o profile.svg --pid <PID>

# 示例代码
import time
import threading

def cpu_bound_task():
    """CPU密集型任务"""
    total = 0
    for i in range(10000000):
        total += i ** 2
    return total

def io_bound_task():
    """IO密集型任务"""
    time.sleep(1)

# 启动多个线程
threads = []
for _ in range(4):
    t = threading.Thread(target=cpu_bound_task)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

性能优化技巧

使用内置函数

import timeit

setup = "data = list(range(10000))"

# 手动求和
manual_time = timeit.timeit(
    'total = 0\nfor x in data:\n    total += x',
    setup=setup,
    number=1000
)

# 内置sum
builtin_time = timeit.timeit(
    'sum(data)',
    setup=setup,
    number=1000
)

print(f"手动求和: {manual_time:.4f}秒")
print(f"内置sum: {builtin_time:.4f}秒")
print(f"内置函数快 {manual_time/builtin_time:.2f} 倍")

使用集合和字典

import timeit

setup = """
data = list(range(10000))
target = set(range(5000, 15000))
"""

# 列表查找
list_time = timeit.timeit(
    '[x for x in data if x in target]',
    setup=setup,
    number=1000
)

# 集合查找
set_time = timeit.timeit(
    'data_set = set(data)\n[x for x in data_set if x in target]',
    setup=setup,
    number=1000
)

print(f"列表查找: {list_time:.4f}秒")
print(f"集合查找: {set_time:.4f}秒")

使用字符串拼接

import timeit

# 字符串拼接方式比较
setup = "words = ['hello', 'world'] * 1000"

concat_time = timeit.timeit(
    "result = ''\nfor word in words:\n    result += word",
    setup=setup,
    number=1000
)

join_time = timeit.timeit(
    "result = ''.join(words)",
    setup=setup,
    number=1000
)

print(f"字符串拼接: {concat_time:.4f}秒")
print(f"join方法: {join_time:.4f}秒")
print(f"join方法快 {concat_time/join_time:.2f} 倍")

性能分析流程

1. 确定目标：明确要优化的代码段
2. 基准测试：使用timeit建立性能基准
3. 分析瓶颈：使用cProfile或line_profiler找到热点
4. 优化代码：根据分析结果进行优化
5. 验证效果：再次测试确认优化有效
6. 持续监控：定期进行性能分析，防止性能退化

性能优化是一个迭代过程，关键是要先分析后优化，避免过早优化。