RSS
菜单
全部文章快讯开发科技深度热点

性能剖析与优化(Python 从精通到入门 · 09)

内容摘要

优化的第一原则不是"写得更快",而是"先测量"。这一篇教你用工具找到真正的瓶颈,然后只改那一处。

优化的第一原则不是"写得更快",而是"先测量"。这一篇教你用工具找到真正的瓶颈,然后只改那一处。

你将学到

  • 为什么"凭直觉优化"几乎总是错的:过早优化是万恶之源
  • 用 timeit 做微基准的正确姿势与常见陷阱
  • 用 cProfile / py-spy / line_profiler 定位热点函数与热点行
  • 用 tracemalloc 揪出内存泄漏和大对象
  • 一批高性价比的优化手段:选对数据结构、缓存、批量 IO、__slots__
  • 一个"优化前 → 剖析 → 优化后"的完整实战案例

前置知识

建议先读 06 - GIL 真相与并发模型选择(性能与并发常常连在一起),本篇收尾整个高手篇。


一、先测量,再优化

Knuth 那句名言要完整地引用:"过早优化是万恶之源"。它的潜台词是——在测量之前,你的直觉有 90% 的概率是错的。

# 三个常见误区:
# 1) 猜瓶颈:"这个循环肯定慢" —— 它可能只占 2% 的时间,改了也没用。
# 2) 凭"看起来优雅"下判断:sorted() 内部是 C 实现的 Timsort,常比手写"聪明算法"还快。
# 3) 优化了错的层级:花一天省下 3ms,却不知道 90% 的时间花在某个数据库查询上。
# 正确流程:定义指标 -> 测量现状 -> 找热点 -> 只优化热点 -> 回到第 2 步验证。

二、微基准:timeit

timeit 解决"单个小操作谁更快",比手写 time.time() 更准(自动多轮运行取最小值、尽量排除干扰)。

import timeit

# 对比两种字符串拼接方式
t1 = timeit.timeit(lambda: "a" + "b" + "c", number=1_000_000)
t2 = timeit.timeit(lambda: "".join(["a", "b", "c"]), number=1_000_000)
print(f"加号拼接: {t1:.4f}s")      # 输出(量级示意): 加号拼接: 0.0500s
print(f"join 拼接: {t2:.4f}s")     # 输出(量级示意): join 拼接: 0.1200s
# 小规模下加号其实更快;join 的优势在大规模、动态切片时才明显

# 命令行版本,不需要写脚本:
#   python -m timeit -s "s='a'" "'x' + s + 'y'"

timeit 的坑:基准环境太干净,脱离真实数据规模和缓存状态,结论常常不能直接搬到生产。它只适合"二选一"的微决策,不适合定位整体瓶颈。

三、定位热点:cProfile / py-spy / line_profiler

cProfile:函数级剖析,标准库自带

import cProfile, pstats, io

def slow_part():
    total = 0
    for i in range(200_000):
        total += i ** 2
    return total

def main():
    for _ in range(20):
        slow_part()
    sum(x for x in range(1000))

pr = cProfile.Profile()
pr.enable(); main(); pr.disable()

s = io.StringIO()
# 按累计耗时排序,打印耗时前 10 的函数
pstats.Stats(pr, stream=s).sort_stats("cumulative").print_stats(10)
print(s.getvalue())
# 输出片段:
#   ncalls  tottime  cumtime  function
#       20    0.850    0.850  perf.py:5(slow_part)   <- 一眼看出它吃掉几乎所有时间
#        1    0.000    0.860  perf.py:13(main)

py-spy:不改代码,直接看运行中的进程

cProfile 需要你改代码、且会给程序加明显开销。py-spy 是采样剖析器,能 attach 到正在运行的进程,适合线上排障。

pip install py-spy

py-spy top --pid 12345                              # 实时看进程在干什么(类似 top)
py-spy record -o profile.svg --pid 12345 --duration 30   # 采火焰图,用浏览器打开
py-spy record -o profile.svg -- python train.py      # 直接剖析脚本

火焰图怎么看:横轴是耗时占比(越宽越热点),纵轴是调用栈。 找最宽的那一摞,就是你的瓶颈。

line_profiler:精确到行

# pip install line_profiler
# 给要剖析的函数加 @profile 装饰器(这是 line_profiler 约定的名字)
@profile
def compute(data):
    result = []
    for x in data:
        result.append(x ** 2 + 2 * x + 1)   # ← 哪一行最贵一目了然
    return sum(result)

compute(list(range(100_000)))
# 命令行运行:kernprof -l -v profile_demo.py
# 输出(示例):
# Line #  Hits      Time  Per Hit   % Time  Line Contents
#      8 100000    85000.0      0.9     95.2      for x in data:

四、内存剖析:tracemalloc

import tracemalloc

tracemalloc.start()          # 开始追踪内存分配

data = [str(i) * 10 for i in range(100_000)]      # 假设这里跑一段可能吃内存的代码
big = {i: [i] * 5 for i in range(50_000)}

snapshot = tracemalloc.take_snapshot()
for stat in snapshot.statistics("lineno")[:3]:    # 按源码行统计,看 Top 3
    print(stat)
# 输出片段:
# perf.py:6: size=9.6 MiB, count=100000, average=101 B
# perf.py:7: size=8.1 MiB, count=50000, average=170 B
# ← 直接告诉你哪一行分配了多少内存、多少个对象

current, peak = tracemalloc.get_traced_memory()
print(f"当前 {current / 1e6:.1f}MB, 峰值 {peak / 1e6:.1f}MB")
tracemalloc.stop()

找内存泄漏的常用手法:取两次快照做 diff。

import tracemalloc

tracemalloc.start()
snap1 = tracemalloc.take_snapshot()

leaky = []
for i in range(10_000):
    leaky.append(object())      # 忘了释放的累积对象

snap2 = tracemalloc.take_snapshot()
for stat in snap2.compare_to(snap1, "lineno")[:3]:
    print(stat)                 # 输出增长最多的行,泄漏点现形

五、高性价比的优化手段

按"投入产出比"从高到低列:

1. 选对数据结构

# ❌ 用列表做成员检查:O(n)
nums = list(range(100_000))
99999 in nums                 # 慢
99999 in set(nums)            # ✅ 转成集合:O(1),快几个数量级

# ❌ 频繁在列表头部插入/删除:O(n);✅ 用 collections.deque 两端操作:O(1)
from collections import deque
dq = deque()
dq.appendleft(1)

2. 把循环推给内置函数 / C 实现

data = list(range(1_000_000))
total = 0
for x in data:                # ❌ Python 层循环
    total += x
total = sum(data)             # ✅ 内置 sum,C 层执行
import numpy as np            # ✅ 更进一步:numpy 向量化(数值场景)
np.arange(1_000_000).sum()

3. 缓存重复计算

from functools import lru_cache
def fib(n):                   # ❌ 每次递归都重算,指数级
    return n if n < 2 else fib(n - 1) + fib(n - 2)

@lru_cache(maxsize=None)      # ✅ 加缓存,指数级降到线性
def fib_cached(n):
    return n if n < 2 else fib_cached(n - 1) + fib_cached(n - 2)
# fib(35) 可能要好几秒;fib_cached(35) 瞬间返回

4. 批量 IO,别一行一次

# ❌ 循环里反复写:每写一次都可能触发系统调用
# with open("out.txt", "w") as f:
#     for x in data: f.write(str(x))
with open("out.txt", "w") as f:            # ✅ 拼成大批次一次写
    f.writelines(f"{x}\n" for x in data)   # 一次性、缓冲写出

5. 用 __slots__ 省内存

当你有几十万个小对象时,__slots__ 能砍掉每实例的 __dict__,节省可观内存。

class Point:
    __slots__ = ("x", "y")     # 禁止动态加属性,省掉 __dict__
    def __init__(self, x, y):
        self.x = x; self.y = y

p = Point(1, 2)
print(p.x)                      # 输出: 1
# p.z = 3                       # 报错 AttributeError,因为没给 z 留槽位
# 100 万个 Point:带 __slots__ 大概能省 40%~50% 内存

六、实战:一次真实的剖析与优化

任务:给定一份用户行为日志,找出"访问次数最多的 3 个用户"。

优化前(直觉版):

import time, random

def top_users_v1(records):
    counts = {}
    for uid in records:
        counts[uid] = counts.get(uid, 0) + 1     # 两次查表
    # 全排序后取前 3
    return sorted(counts.items(), key=lambda kv: kv[1], reverse=True)[:3]

records = [random.randint(0, 999) for _ in range(1_000_000)]   # 100 万条日志
t0 = time.perf_counter(); print(top_users_v1(records))
print(f"v1 耗时: {time.perf_counter() - t0:.3f}s")   # 输出: v1 耗时: 0.420s

# 用 cProfile 剖析(sort="cumulative"),会看到:
#        1    0.070    0.220  <built-in sorted>   <- 对 1000 个 key 全排序,浪费
#  1000000    0.120    0.120  dict.get / __contains__

优化后(按剖析结论改):

from collections import Counter
import heapq

def top_users_v2(records):
    counts = Counter(records)                                       # 1) C 层计数
    return heapq.nlargest(3, counts.items(), key=lambda kv: kv[1])   # 2) 堆取 top-3,O(n log k)

t0 = time.perf_counter(); print(top_users_v2(records))
print(f"v2 耗时: {time.perf_counter() - t0:.3f}s")   # 输出: v2 耗时: 0.075s

收益:约 5.6 倍。注意——每一处改动都对应剖析里的一个热点(Counter 解决循环计数,heapq 解决全排序),不是"凭感觉堆技巧"。这就是"测量 → 定位 → 只改热点"的正确姿势。

再往前一步,如果记录是流式的、内存吃紧,可以用 Counter.update() 边读边计,或直接写 SQL GROUP BY 交给数据库——优化到某个点,换一个层级(算法 / 存储引擎)比抠 Python 代码收益更大。

常见坑

# ❌ 坑 1:用 time.time() 测微基准,误差比目标还大
import time
t0 = time.time(); x = 1 + 1            # 纳秒级操作
print(time.time() - t0)               # 输出可能是 0.0,测了个寂寞
# ✅ 正解:timeit 自动多轮、取最小,排除干扰
import timeit
print(timeit.timeit("1 + 1", number=1_000_000))
# ❌ 坑 2:在循环里做"看起来很小"的重复计算
def has_permission(user):
    return user is not None and user.role in get_all_roles()   # 每次都重建角色集
# ✅ 正解:把不变量提到循环外(或缓存)
ALL_ROLES = frozenset({"admin", "editor", "viewer"})
def has_permission_v2(user):
    return user is not None and user.role in ALL_ROLES
# ❌ 坑 3:为了"优雅"把 O(n) 结构用成 O(n²)
matches = [a for a in list_a if a in list_b]   # 每次 O(len(list_b)),整体 O(n*m)
# ✅ 正解:把被查找的一方转成 set,整体 O(n + m)
set_b = set(list_b)
matches = [a for a in list_a if a in set_b]

# ❌ 坑 4:优化了没测量的地方("我把这个日志函数重写了,肯定快",
#   而它根本没进剖析结果前 20 名,时间其实花在某个隐式的 json 反序列化上)
# ✅ 正解:让剖析器告诉你热点,改完再测一次确认

小结

  • 先测量再优化:没有数字的优化都是猜测,过早优化是万恶之源。
  • 微基准用 timeit;函数级热点用 cProfile;运行中进程用 py-spy;精确到行用 line_profiler;内存用 tracemalloc。
  • 万能三招:选对数据结构(set/deque 替代 list 的线性操作)、下推到 C(内置函数 / numpy)、缓存(lru_cache)。
  • 批量化 IO,减少系统调用次数;海量小对象考虑 __slots__。
  • 实战案例:Counter + heapq.nlargest 替代"手数 + 全排序",约 5.6 倍提升——每处改动都对应一个已定位的热点。
  • 优化到头时,换层级(算法、数据库、C 扩展)比抠 Python 语法收益更大。

延伸阅读

  • Python 官方文档:timeit、cProfile、tracemalloc 模块
  • py-spy 项目主页(火焰图与采样剖析)
  • 《High Performance Python》(Micha Gorelick & Ian Ozsvald)
— 全文完 —回到顶部 ↑
下载推广海报

文章推广海报

《性能剖析与优化(Python 从精通到入门 · 09)》完整推广海报
DISCUSSION

文章回复

0 条公开回复
未登录回复需要审核后公开
还没有回复,欢迎参与讨论。