python-profiling

Python 性能分析

不适用此技能的情况

• Java/JVM 性能分析 - 使用 java-profiling 技能进行 JFR 和 GC 调优
• Node.js 性能分析 - 使用 nodejs-profiling 技能进行 V8 性能分析
• NumPy/Pandas 优化 - 使用库专用的性能分析工具和向量化指南
• 数据库查询优化 - 使用数据库专用的性能分析工具
• Web 服务器性能 - 使用应用级性能分析（Django Debug Toolbar、Flask-DebugToolbar）

深入了解：使用 mcp__documentation__fetch_docs，并将 technology 设为 python，获取全面的性能分析指南、优化技巧和最佳实践。

cProfile（CPU 性能分析）

命令行用法

# Profile entire script
python -m cProfile -o output.prof script.py

# Sort by cumulative time
python -m cProfile -s cumtime script.py

# Sort by total time in function
python -m cProfile -s tottime script.py

# Analyze saved profile
python -m pstats output.prof

pstats 分析

import pstats

# Load and analyze profile
stats = pstats.Stats('output.prof')
stats.strip_dirs()
stats.sort_stats('cumulative')
stats.print_stats(20)  # Top 20 functions

# Filter by module
stats.print_stats('mymodule')

# Show callers
stats.print_callers('slow_function')

# Show callees
stats.print_callees('main')

编程方式进行性能分析

import cProfile
import pstats
from io import StringIO

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

    result = func(*args, **kwargs)

    profiler.disable()

    # Analyze
    stream = StringIO()
    stats = pstats.Stats(profiler, stream=stream)
    stats.sort_stats('cumulative')
    stats.print_stats(10)
    print(stream.getvalue())

    return result

# Context manager
from contextlib import contextmanager

@contextmanager
def profile_block(name='profile'):
    profiler = cProfile.Profile()
    profiler.enable()
    try:
        yield
    finally:
        profiler.disable()
        profiler.dump_stats(f'{name}.prof')

内存分析

tracemalloc（内置）

import tracemalloc

# Start tracking
tracemalloc.start()

# Your code here
result = process_data()

# Get snapshot
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

print("Top 10 memory allocations:")
for stat in top_stats[:10]:
    print(stat)

# Compare snapshots
snapshot1 = tracemalloc.take_snapshot()
# ... code ...
snapshot2 = tracemalloc.take_snapshot()

diff = snapshot2.compare_to(snapshot1, 'lineno')
for stat in diff[:10]:
    print(stat)

# Stop tracking
tracemalloc.stop()

memory_profiler（逐行分析）

# Install: pip install memory_profiler

from memory_profiler import profile

@profile
def my_function():
    a = [1] * 1_000_000
    b = [2] * 2_000_000
    del b
    return a

# Command line usage
# python -m memory_profiler script.py

# Profile specific function
# mprof run script.py
# mprof plot

objgraph（对象引用）

# Install: pip install objgraph

import objgraph

# Most common types
objgraph.show_most_common_types(limit=20)

# Growth since last call
objgraph.show_growth()

# Find reference chain (memory leak detection)
objgraph.show_backrefs([leaked_object], filename='refs.png')

行性能分析器

# Install: pip install line_profiler

# Decorate functions to profile
@profile
def slow_function():
    total = 0
    for i in range(1000000):
        total += i
    return total

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

高精度计时

time 模块

import time

# Monotonic clock (best for measuring durations)
start = time.perf_counter()
result = do_work()
duration = time.perf_counter() - start
print(f"Duration: {duration:.4f}s")

# Nanosecond precision (Python 3.7+)
start = time.perf_counter_ns()
result = do_work()
duration_ns = time.perf_counter_ns() - start
print(f"Duration: {duration_ns}ns")

timeit 模块

import timeit

# Time small code snippets
duration = timeit.timeit('sum(range(1000))', number=10000)
print(f"Average: {duration / 10000:.6f}s")

# Compare implementations
setup = "data = list(range(10000))"
time1 = timeit.timeit('sum(data)', setup, number=1000)
time2 = timeit.timeit('sum(x for x in data)', setup, number=1000)
print(f"sum(): {time1:.4f}s, generator: {time2:.4f}s")

常见瓶颈模式

列表操作

# ❌ Bad: Concatenating lists in loop
result = []
for item in items:
    result = result + [process(item)]  # O(n²)

# ✅ Good: Use append
result = []
for item in items:
    result.append(process(item))  # O(n)

# ✅ Better: List comprehension
result = [process(item) for item in items]

# ❌ Bad: Checking membership in list
if item in large_list:  # O(n)
    pass

# ✅ Good: Use set for membership
large_set = set(large_list)
if item in large_set:  # O(1)
    pass

字符串操作

# ❌ Bad: String concatenation in loop
result = ""
for s in strings:
    result += s  # Creates new string each time

# ✅ Good: Use join
result = "".join(strings)

# ❌ Bad: Format in loop
for item in items:
    log(f"Processing {item}")

# ✅ Good: Lazy formatting
import logging
for item in items:
    logging.debug("Processing %s", item)  # Only formats if needed

字典操作

# ❌ Bad: Repeated key lookup
if key in d:
    value = d[key]
    process(value)

# ✅ Good: Use get or setdefault
value = d.get(key)
if value is not None:
    process(value)

# ❌ Bad: Checking then setting
if key not in d:
    d[key] = []
d[key].append(value)

# ✅ Good: Use defaultdict
from collections import defaultdict
d = defaultdict(list)
d[key].append(value)

生成器与列表

# ❌ Bad: Creating large intermediate lists
result = sum([x * 2 for x in range(10_000_000)])  # Uses memory

# ✅ Good: Use generator
result = sum(x * 2 for x in range(10_000_000))  # Lazy evaluation

# Process large files
# ❌ Bad
data = open('large.csv').readlines()  # All in memory
for line in data:
    process(line)

# ✅ Good
with open('large.csv') as f:  # Stream line by line
    for line in f:
        process(line)

NumPy 优化

import numpy as np

# ❌ Bad: Python loops over arrays
result = []
for i in range(len(arr)):
    result.append(arr[i] * 2)

# ✅ Good: Vectorized operations
result = arr * 2  # SIMD operations

# ❌ Bad: Creating many temporary arrays
result = (arr1 + arr2) * arr3 / arr4  # 3 temporaries

# ✅ Good: In-place operations when possible
result = arr1.copy()
result += arr2
result *= arr3
result /= arr4

# Use appropriate dtypes
arr = np.array(data, dtype=np.float32)  # Half memory of float64

异步优化

import asyncio
import aiohttp

# ❌ Bad: Sequential async
async def fetch_all_sequential(urls):
    results = []
    async with aiohttp.ClientSession() as session:
        for url in urls:
            async with session.get(url) as resp:
                results.append(await resp.text())
    return results

# ✅ Good: Concurrent async
async def fetch_all_concurrent(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [await r.text() for r in responses]

# ✅ Better: With concurrency limit
from asyncio import Semaphore

async def fetch_with_limit(urls, limit=10):
    semaphore = Semaphore(limit)

    async def fetch_one(url):
        async with semaphore:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as resp:
                    return await resp.text()

    return await asyncio.gather(*[fetch_one(url) for url in urls])

多进程

from multiprocessing import Pool, cpu_count
from concurrent.futures import ProcessPoolExecutor

# CPU-bound work
def cpu_intensive(x):
    return sum(i * i for i in range(x))

# Using Pool
with Pool(cpu_count()) as pool:
    results = pool.map(cpu_intensive, range(100))

# Using ProcessPoolExecutor
with ProcessPoolExecutor() as executor:
    results = list(executor.map(cpu_intensive, range(100)))

# Shared memory (Python 3.8+)
from multiprocessing import shared_memory
import numpy as np

# Create shared array
shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
shared_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
shared_arr[:] = arr[:]

性能分析检查清单

检查项	工具	命令
CPU 热点	cProfile	python -m cProfile script.py
逐行分析	line_profiler	kernprof -l -v script.py
内存用量	tracemalloc	tracemalloc.start()
每行内存用量	memory_profiler	@profile 装饰器
对象引用	objgraph	objgraph.show_growth()
快速基准测试	timeit	timeit.timeit()

py-spy（采样式性能分析器）

# Install: pip install py-spy

# Record profile
py-spy record -o profile.svg -- python script.py

# Top-like view of running process
py-spy top --pid <pid>

# Dump current stack
py-spy dump --pid <pid>

# Profile subprocesses
py-spy record --subprocesses -o profile.svg -- python script.py

生产环境优化

# Use __slots__ for memory efficiency
class Point:
    __slots__ = ['x', 'y']
    def __init__(self, x, y):
        self.x = x
        self.y = y

# Use lru_cache for memoization
from functools import lru_cache

@lru_cache(maxsize=1000)
def expensive_computation(x):
    return x ** 2

# Use dataclasses with slots (Python 3.10+)
from dataclasses import dataclass

@dataclass(slots=True)
class Point:
    x: float
    y: float

反模式

反模式	错误原因	正确做法
在循环中使用 + 拼接字符串	时间复杂度为 O(n²)	使用 ''.join() 或列表推导式
能用生成器时却使用列表推导式	造成不必要的内存分配	一次性迭代时使用生成器表达式
需要 enumerate() 时却使用 range()	手动跟踪索引，容易出错	使用 enumerate() 获取索引和值
检查元素是否在列表中	查找复杂度为 O(n)	使用 set，以 O(1) 复杂度检查成员关系
到处使用 global 变量	难以进行性能分析，且会产生副作用	传递参数并返回值
数值运算时不使用 NumPy	速度会慢几个数量级	使用 NumPy 对数组运算进行向量化
过早优化	浪费精力，增加维护难度	先进行性能分析，再优化瓶颈
使用 import *	污染命名空间，导入速度较慢	导入指定名称
已知列表大小时仍在循环中使用 .append()	导致多次重新分配	使用列表推导式或 [None] * size 预先分配
大量实例未使用 __slots__	内存用量更高	为实例众多的类使用 __slots__

快速故障排除

问题	诊断	解决方案
遍历大型数据时速度缓慢	Python 循环速度较慢	使用 NumPy 向量化，或使用列表推导式
内存使用量过高	创建了大型中间对象	使用生成器，分块处理
GIL 争用	多线程无法加快 CPU 密集型任务	对 CPU 密集型任务使用 multiprocessing
导入速度缓慢	模块较大且包含副作用	延迟导入，减少模块级代码
内存泄漏	对象未被垃圾回收	检查循环引用，使用 weakref
RecursionError	递归层数过深	使用 sys.setrecursionlimit() 提高限制，或改为迭代
字典操作速度缓慢	哈希冲突	确保键可哈希且分布均匀
分析器中 CPU 占用率高	未显示 C 扩展	使用 py-spy 等采样分析器
处理大型文件时内存不足	加载了整个文件	使用 with open() 并逐行迭代
JSON 解析速度缓慢	JSON 文件较大	使用流式解析器（ijson）或 pandas

相关技能

• FastAPI
• Django
• NumPy/Pandas