Python Performance

Python 性能优化

概述

掌握 Python 性能优化。学习如何分析代码性能、找出瓶颈、优化算法、高效管理内存，并利用高性能库处理计算密集型任务。

学习目标

• 分析 Python 代码性能，找出瓶颈
• 优化算法和数据结构
• 高效管理内存
• 使用编译扩展（Cython、NumPy）
• 实现缓存策略
• 并行处理 CPU 密集型操作
• 对改进效果进行基准测试和测量

核心主题

1. 性能分析与基准测试

• 使用 timeit 模块进行微基准测试
• 使用 cProfile 进行函数级性能分析
• 使用 line_profiler 逐行分析
• 使用 memory_profiler 分析内存使用情况
• 使用 py-spy 分析生产环境性能
• 火焰图和可视化

代码示例：

import timeit
import cProfile
import pstats

# 1. timeit for micro-benchmarks
def list_comprehension():
    return [x**2 for x in range(1000)]

def map_function():
    return list(map(lambda x: x**2, range(1000)))

# Compare performance
time_lc = timeit.timeit(list_comprehension, number=10000)
time_map = timeit.timeit(map_function, number=10000)
print(f"List comprehension: {time_lc:.4f}s")
print(f"Map function: {time_map:.4f}s")

# 2. cProfile for function profiling
def process_data():
    data = []
    for i in range(100000):
        data.append(i ** 2)
    return sum(data)

profiler = cProfile.Profile()
profiler.enable()
result = process_data()
profiler.disable()

stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)

# 3. Line profiling (requires line_profiler package)
# @profile decorator (add manually for line_profiler)
def slow_function():
    total = 0
    for i in range(1000000):
        total += i ** 2
    return total

# Run with: kernprof -l -v script.py

# 4. Memory profiling
from memory_profiler import profile

@profile
def memory_intensive():
    large_list = [i for i in range(1000000)]
    large_dict = {i: i**2 for i in range(1000000)}
    return len(large_list) + len(large_dict)

# Run with: python -m memory_profiler script.py

2. 算法与数据结构优化

• 选择高效的数据结构
• 时间复杂度分析
• 生成器表达式与列表
• 使用集合操作进行查找
• 使用 Deque 执行队列操作
• 使用 Bisect 处理已排序列表

代码示例：

import bisect
from collections import deque, Counter, defaultdict
import time

# 1. List vs Set for membership testing
# Bad: O(n) lookup
def find_in_list(items, target):
    return target in items  # Linear search

# Good: O(1) lookup
def find_in_set(items, target):
    items_set = set(items)
    return target in items_set

items = list(range(100000))
# List: 0.001s, Set: 0.000001s (1000x faster!)

# 2. Generator expressions for memory efficiency
# Bad: Creates entire list in memory
squares_list = [x**2 for x in range(1000000)]  # ~4MB

# Good: Generates on-demand
squares_gen = (x**2 for x in range(1000000))   # ~128 bytes

# 3. Deque for efficient queue operations
# Bad: O(n) pop from beginning
queue_list = list(range(10000))
queue_list.pop(0)  # Slow

# Good: O(1) pop from both ends
queue_deque = deque(range(10000))
queue_deque.popleft()  # Fast

# 4. Bisect for maintaining sorted lists
# Bad: O(n) insertion into sorted list
sorted_list = []
for i in [5, 2, 8, 1, 9]:
    sorted_list.append(i)
    sorted_list.sort()

# Good: O(log n) insertion
sorted_list = []
for i in [5, 2, 8, 1, 9]:
    bisect.insort(sorted_list, i)

# 5. Counter for frequency counting
# Bad: Manual counting
word_count = {}
for word in words:
    if word in word_count:
        word_count[word] += 1
    else:
        word_count[word] = 1

# Good: Counter
word_count = Counter(words)
most_common = word_count.most_common(10)

3. 内存管理

• 内存分配和垃圾回收
• 对象池
• 使用 Slots 创建节省内存的类
• 引用计数
• 弱引用
• 内存泄漏检测

代码示例：

import gc
import sys
from weakref import WeakValueDictionary

# 1. __slots__ for memory-efficient classes
# Bad: Regular class (56 bytes per instance)
class RegularPoint:
    def __init__(self, x, y):
        self.x = x
        self.y = y

# Good: Slots class (32 bytes per instance - 43% smaller!)
class SlottedPoint:
    __slots__ = ['x', 'y']

    def __init__(self, x, y):
        self.x = x
        self.y = y

print(sys.getsizeof(RegularPoint(1, 2)))  # 56 bytes
print(sys.getsizeof(SlottedPoint(1, 2)))  # 32 bytes

# 2. Object pooling for expensive objects
class ObjectPool:
    def __init__(self, factory, max_size=10):
        self.factory = factory
        self.max_size = max_size
        self.pool = []

    def acquire(self):
        if self.pool:
            return self.pool.pop()
        return self.factory()

    def release(self, obj):
        if len(self.pool) < self.max_size:
            self.pool.append(obj)

# Usage
db_pool = ObjectPool(lambda: DatabaseConnection(), max_size=5)
conn = db_pool.acquire()
# Use connection
db_pool.release(conn)

# 3. Weak references to prevent memory leaks
class Cache:
    def __init__(self):
        self._cache = WeakValueDictionary()

    def get(self, key):
        return self._cache.get(key)

    def set(self, key, value):
        self._cache[key] = value

# 4. Manual garbage collection for large operations
def process_large_dataset():
    for batch in large_data:
        process_batch(batch)
        # Force garbage collection after each batch
        gc.collect()

# 5. Context managers for resource cleanup
class ManagedResource:
    def __enter__(self):
        self.resource = allocate_resource()
        return self.resource

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.resource.cleanup()
        return False

4. 高性能计算

• NumPy 向量化
• Numba JIT 编译
• 使用 Cython 编写 C 扩展
• 使用多进程实现并行处理
• Concurrent.futures
• 性能比较

代码示例：

import numpy as np
from numba import jit
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor

# 1. NumPy vectorization
# Bad: Python loops (slow)
def python_sum(n):
    total = 0
    for i in range(n):
        total += i ** 2
    return total

# Good: NumPy vectorization (100x faster!)
def numpy_sum(n):
    arr = np.arange(n)
    return np.sum(arr ** 2)

# Benchmark: python_sum(1000000) = 0.15s
#           numpy_sum(1000000)  = 0.002s

# 2. Numba JIT compilation
@jit(nopython=True)  # Compile to machine code
def fast_function(n):
    total = 0
    for i in range(n):
        total += i ** 2
    return total

# First call: compilation + execution
# Subsequent calls: 50x faster than pure Python!

# 3. Multiprocessing for CPU-bound tasks
def cpu_intensive_task(n):
    return sum(i * i for i in range(n))

# Single process
result = cpu_intensive_task(10000000)

# Multiple processes
with ProcessPoolExecutor(max_workers=4) as executor:
    ranges = [2500000, 2500000, 2500000, 2500000]
    results = executor.map(cpu_intensive_task, ranges)
    total = sum(results)

# 4x speedup on 4 cores!

# 4. Caching for expensive computations
from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# fibonacci(100) without cache: ~forever
# fibonacci(100) with cache: instant

# 5. Memory views for zero-copy operations
def process_array(data):
    # Bad: Creates copy
    subset = data[1000:2000]

    # Good: Zero-copy view
    view = memoryview(data)[1000:2000]

动手实践

项目 1：性能分析工具

构建一个全面的性能分析工具。

要求：

• 使用 cProfile 分析 CPU 性能
• 内存性能分析
• 逐行分析
• 可视化（火焰图）
• 生成 HTML 报告
• 识别瓶颈

核心技能： 性能分析工具、可视化、分析

项目 2：数据处理流水线

优化数据处理流水线。

要求：

• 加载大型 CSV 文件（1GB 以上）
• 转换和清理数据
• 汇总统计数据
• 比较 Python、NumPy 和 Pandas 方案
• 测量内存使用情况
• 将内存占用优化至 <2GB

核心技能： NumPy、内存优化、基准测试

项目 3：并行计算

实现并行算法。

要求：

• 矩阵乘法
• 图像处理
• 蒙特卡罗模拟
• 比较多线程、多进程和 asyncio
• 测量加速比
• 处理共享状态

核心技能： 并行处理、性能测量

评估标准

• 分析代码性能，找出瓶颈
• 选择合适的数据结构
• 优化算法的时间复杂度
• 高效管理内存
• 在适用场景使用向量化
• 实现有效的缓存
• 并行处理 CPU 密集型操作

资源

官方文档

• Python 性能技巧 - 官方技巧
• NumPy 文档 - NumPy 文档
• Numba 文档 - JIT 编译

学习平台

• 高性能 Python - O'Reilly 图书
• Python 性能 - Real Python 指南
• 优化 Python - PyCon 演讲

工具

• cProfile - CPU 性能分析
• memory_profiler - 内存性能分析
• py-spy - 采样分析器
• Scalene - CPU/GPU/内存性能分析器

后续步骤

掌握 Python 性能优化后，可以进一步探索：

• Cython - Python 的 C 扩展
• PyPy - 替代 Python 解释器
• Dask - 并行计算库
• CUDA - 使用 Python 进行 GPU 编程