Python 性能优化实战指南：cProfile 瓶颈定位 + __slots__ 内存优化 + NumPy 向量化 + functools.cache

1
import cProfile
2
import pstats
3
import io
4

5
def slow_function():
6
    result = []
7
    for i in range(100_000):
8
        result.append(i ** 2)
9
    return result
10

11
# 方法一：直接 run
12
cProfile.run('slow_function()', sort='cumulative')
13

14
# 方法二：更精细的控制
15
def profile_this(func, *args, **kwargs):
16
    pr = cProfile.Profile()
17
    pr.enable()
18
    result = func(*args, **kwargs)
19
    pr.disable()
20

21
    stream = io.StringIO()
22
    ps = pstats.Stats(pr, stream=stream).sort_stats('cumulative')
23
    ps.print_stats(15)  # 只显示最慢的 15 个函数
24
    print(stream.getvalue())
25
    return result
26

27
profile_this(slow_function)

1
         200003 function calls in 0.087 seconds
2

3
   Ordered by: cumulative time
4

5
   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
6
        1    0.000    0.000    0.087    0.087 <string>:1(<module>)
7
        1    0.041    0.041    0.087    0.087 script.py:4(slow_function)
8
   100000    0.046    0.000    0.046    0.000 {method 'append' of 'list'}

1
# 用 @profile 装饰器标记要分析的函数（line_profiler 注入）
2
@profile
3
def process_data(data):
4
    result = []
5
    for item in data:           # ← 想知道这行占多少时间？
6
        cleaned = item.strip()
7
        if cleaned:
8
            result.append(cleaned.upper())
9
    return result
10

11
data = [f"  item_{i}  " for i in range(100_000)]
12
process_data(data)

1
Line #   Hits    Time   Per Hit  % Time  Line Contents
2
================================================
3
     5  100000  15234.0    0.2   18.3    for item in data:
4
     6  100000  22187.0    0.2   26.6        cleaned = item.strip()
5
     7  100000   9823.0    0.1   11.8        if cleaned:
6
     8  100000  36247.0    0.4   43.4            result.append(cleaned.upper())

1
import timeit
2

3
# 对比两种字符串连接方式
4
code_a = """
5
result = ""
6
for i in range(1000):
7
    result += str(i)
8
"""
9

10
code_b = """
11
result = "".join(str(i) for i in range(1000))
12
"""
13

14
time_a = timeit.timeit(code_a, number=1000)
15
time_b = timeit.timeit(code_b, number=1000)
16

17
print(f"字符串 += ：{time_a:.4f}s")
18
print(f"join()    ：{time_b:.4f}s")
19
print(f"join() 快了 {time_a / time_b:.1f}x")

1
字符串 += ：0.2847s
2
join()    ：0.0521s
3
join() 快了 5.5x

1
import sys
2

3
class Point:
4
    def __init__(self, x, y, z):
5
        self.x = x
6
        self.y = y
7
        self.z = z
8

9
p = Point(1.0, 2.0, 3.0)
10
print(sys.getsizeof(p))           # 约 48 字节（对象本身）
11
print(sys.getsizeof(p.__dict__))  # 约 232 字节（属性字典！）
12
# 合计：约 280 字节/实例

1
class PointSlots:
2
    __slots__ = ('x', 'y', 'z')   # 声明允许的属性名
3

4
    def __init__(self, x, y, z):
5
        self.x = x
6
        self.y = y
7
        self.z = z
8

9
ps = PointSlots(1.0, 2.0, 3.0)
10
print(sys.getsizeof(ps))          # 约 72 字节
11
# 无 __dict__！节省约 75%

1
import tracemalloc
2

3
def benchmark_memory(cls, n=1_000_000):
4
    tracemalloc.start()
5
    objects = [cls(i * 1.0, i * 2.0, i * 3.0) for i in range(n)]
6
    current, peak = tracemalloc.get_traced_memory()
7
    tracemalloc.stop()
8
    return peak / 1024 / 1024  # 转换为 MB
9

10
print(f"普通 class：{benchmark_memory(Point):.1f} MB")
11
print(f"__slots__  ：{benchmark_memory(PointSlots):.1f} MB")

1
普通 class：258.4 MB
2
__slots__  ：76.3 MB    # 节省 70%+

1
# ❌ 不能动态添加属性
2
ps = PointSlots(1, 2, 3)
3
ps.w = 4  # → AttributeError: 'PointSlots' object has no attribute 'w'
4

5
# ❌ 继承时需要子类也声明 __slots__，否则子类会恢复 __dict__
6
class ColorPoint(PointSlots):
7
    __slots__ = ('color',)   # 继承类也要声明！
8
    def __init__(self, x, y, z, color):
9
        super().__init__(x, y, z)
10
        self.color = color
11

12
# ✅ 适用场景：大量创建的小数据对象（记录、坐标、事件等）
13
# ❌ 不适合：需要动态属性、pickle 序列化复杂场景、多重继承

1
from dataclasses import dataclass
2

3
@dataclass(slots=True)   # Python 3.10+ 一行搞定
4
class Point3D:
5
    x: float
6
    y: float
7
    z: float

1
Python for 循环：
2
  每次迭代 → 解释器调度字节码 → 类型检查 → 装箱/拆箱 → 内存分配
3
  100 万次迭代 = 100 万次以上的解释器操作
4

5
NumPy 向量化：
6
  一次 C 函数调用 → SIMD 指令批量处理 → 连续内存读写
7
  无 Python 解释器参与，数据在 C 层批量处理

1
import numpy as np
2
import time
3

4
N = 1_000_000
5

6
# ── 场景 1：计算平方和 ─────────────────────────────────────────
7
def pure_python_sum_squares(n):
8
    return sum(i ** 2 for i in range(n))
9

10
def numpy_sum_squares(n):
11
    arr = np.arange(n, dtype=np.float64)
12
    return np.sum(arr ** 2)
13

14
# 性能对比
15
t0 = time.perf_counter()
16
result_py = pure_python_sum_squares(N)
17
t1 = time.perf_counter()
18
result_np = numpy_sum_squares(N)
19
t2 = time.perf_counter()
20

21
print(f"纯 Python：{t1 - t0:.4f}s")
22
print(f"NumPy    ：{t2 - t1:.4f}s")
23
print(f"加速比   ：{(t1 - t0) / (t2 - t1):.1f}x")
24
# 典型输出：NumPy 快约 50-80x

1
import numpy as np
2

3
# ── 条件筛选（替代 if-for 组合）──────────────────────────────
4
data = np.random.randn(1_000_000)
5

6
# ❌ 慢：Python for + if
7
result = [x for x in data if x > 0]
8

9
# ✅ 快：布尔索引
10
result = data[data > 0]
11

12

13
# ── 逐元素运算（替代 map/for）────────────────────────────────
14
# ❌ 慢
15
prices = [100.0, 200.0, 150.0, 300.0]
16
discounted = [p * 0.9 for p in prices]
17

18
# ✅ 快
19
prices_np = np.array(prices)
20
discounted_np = prices_np * 0.9   # 广播运算，一次完成
21

22

23
# ── 矩阵运算（替代嵌套 for 循环）────────────────────────────
24
# ❌ 极慢：O(n³) 嵌套循环
25
def matrix_mult_python(A, B):
26
    n = len(A)
27
    C = [[0.0] * n for _ in range(n)]
28
    for i in range(n):
29
        for j in range(n):
30
            for k in range(n):
31
                C[i][j] += A[i][k] * B[k][j]
32
    return C
33

34
# ✅ 极快：NumPy 调用 BLAS 底层
35
A = np.random.rand(500, 500)
36
B = np.random.rand(500, 500)
37
C = A @ B   # 或 np.matmul(A, B)
38
# 500x500 矩阵：Python ≈ 60s，NumPy ≈ 0.01s，快约 6000x
39

40

41
# ── 字符串/分类数据：用 np.where 替代 if-else ────────────────
42
scores = np.array([85, 42, 96, 60, 73])
43
grades = np.where(scores >= 60, 'pass', 'fail')
44

45

46
# ── 聚合统计（替代 statistics 模块）──────────────────────────
47
data = np.random.randn(1_000_000)
48
mean   = np.mean(data)    # 均值
49
std    = np.std(data)     # 标准差
50
median = np.median(data)  # 中位数
51
pct95  = np.percentile(data, 95)  # 95 分位数

1
import numpy as np
2

3
# ── 陷阱1：在循环中访问 NumPy 数组元素（退化为 Python 速度）──
4
arr = np.arange(1_000_000, dtype=np.float64)
5

6
# ❌ 错误：逐元素访问，失去向量化优势
7
total = 0
8
for x in arr:           # 每次访问都有 Python 对象转换开销
9
    total += x
10

11
# ✅ 正确：一次向量化调用
12
total = arr.sum()
13

14

15
# ── 陷阱2：不必要的数组拷贝 ──────────────────────────────────
16
a = np.arange(1_000_000)
17

18
# ❌ 切片后修改，某些操作会触发拷贝
19
b = a[::2]              # 视图（view），不拷贝
20
b += 1                  # 修改 b 同时修改 a（共享内存）
21

22
c = a[::2].copy()       # 显式拷贝，独立数据
23
c += 1                  # 不影响 a
24

25
# 检查是否是视图
26
print(b.base is a)      # True  → 视图
27
print(c.base is a)      # False → 独立拷贝
28

29

30
# ── 陷阱3：dtype 选择不当导致精度损失或内存浪费 ──────────────
31
# 如果数值范围在 0-255，用 uint8 而不是 float64
32
images = np.zeros((1000, 224, 224, 3), dtype=np.float64)  # 1.2 GB
33
images = np.zeros((1000, 224, 224, 3), dtype=np.uint8)    # 150 MB

1
from functools import cache, lru_cache
2
import time
3

4
# ── 无缓存的递归斐波那契（指数级复杂度）──────────────────────
5
def fib_naive(n):
6
    if n <= 1:
7
        return n
8
    return fib_naive(n - 1) + fib_naive(n - 2)
9

10
# ── @cache：无限制缓存（Python 3.9+）─────────────────────────
11
@cache
12
def fib_cached(n):
13
    if n <= 1:
14
        return n
15
    return fib_cached(n - 1) + fib_cached(n - 2)
16

17
# ── @lru_cache：有 maxsize 限制的 LRU 缓存 ──────────────────
18
@lru_cache(maxsize=256)   # 最多缓存 256 个不同参数的结果
19
def fib_lru(n):
20
    if n <= 1:
21
        return n
22
    return fib_lru(n - 1) + fib_lru(n - 2)
23

24
# 性能对比
25
t0 = time.perf_counter(); fib_naive(35);  t1 = time.perf_counter()
26
t2 = time.perf_counter(); fib_cached(35); t3 = time.perf_counter()
27

28
print(f"无缓存 fib(35)：{t1 - t0:.4f}s")
29
print(f"@cache fib(35)：{t3 - t2:.6f}s")
30
# 无缓存：约 3.5s；@cache：约 0.00001s，快 350,000x

1
# @lru_cache 的缓存统计
2
from functools import lru_cache
3

4
@lru_cache(maxsize=128)
5
def compute(n):
6
    return sum(range(n))
7

8
for i in range(200):
9
    compute(i % 100)   # 循环访问，触发 LRU
10

11
info = compute.cache_info()
12
print(f"命中：{info.hits}，未命中：{info.misses}，当前缓存量：{info.currsize}")
13
# 输出：命中：100，未命中：100，当前缓存量：100
14

15
# 手动清除缓存（如数据更新后）
16
compute.cache_clear()

1
from functools import lru_cache
2
import time
3

4
# 模拟慢速数据库查询
5
def db_query(user_id: int) -> dict:
6
    time.sleep(0.1)   # 模拟 100ms 数据库延迟
7
    return {"id": user_id, "name": f"User{user_id}", "score": user_id * 10}
8

9
# ✅ 缓存热点用户查询（maxsize 根据内存预算和用户活跃度调整）
10
@lru_cache(maxsize=1000)
11
def get_user_cached(user_id: int) -> dict:
12
    return db_query(user_id)
13

14
# 首次查询（冷缓存）
15
t0 = time.perf_counter()
16
user = get_user_cached(42)
17
print(f"首次查询：{time.perf_counter() - t0:.3f}s")  # 约 0.100s
18

19
# 再次查询（缓存命中）
20
t0 = time.perf_counter()
21
user = get_user_cached(42)
22
print(f"缓存查询：{time.perf_counter() - t0:.6f}s")  # 约 0.000001s

1
import timeit
2

3
n = 10_000
4

5
# ❌ 慢：+= 每次创建新字符串对象
6
def concat_plus(n):
7
    s = ""
8
    for i in range(n):
9
        s += str(i)
10
    return s
11

12
# ✅ 快：join 一次性拼接
13
def concat_join(n):
14
    return "".join(str(i) for i in range(n))
15

16
# ✅ 更快：先 map 再 join（避免生成器开销）
17
def concat_join_map(n):
18
    return "".join(map(str, range(n)))
19

20
t_plus     = timeit.timeit(lambda: concat_plus(n),     number=100)
21
t_join     = timeit.timeit(lambda: concat_join(n),     number=100)
22
t_join_map = timeit.timeit(lambda: concat_join_map(n), number=100)
23

24
print(f"+=       : {t_plus:.4f}s")
25
print(f"join     : {t_join:.4f}s  ({t_plus/t_join:.1f}x 提速)")
26
print(f"join+map : {t_join_map:.4f}s  ({t_plus/t_join_map:.1f}x 提速)")
27
# +=：约 0.45s；join+map：约 0.06s，快约 7x

1
name, value = "World", 42
2

3
# 性能（由快到慢）：
4
result_f   = f"Hello {name}, value={value}"         # ✅ 最快（Python 3.6+）
5
result_fmt = "Hello {}, value={}".format(name, value)  # 中等
6
result_pct = "Hello %s, value=%d" % (name, value)     # 最慢（古老写法）
7

8
# f-string 比 .format() 快约 35-40%，是推荐写法