python

《Python编程实战》16.2 CPU·内存·I/O 三类瓶颈的定位与优化

把瓶颈分成 CPU、内存、I/O 三类逐一击破：用 numpy 向量化与算法改进实测加速比，用 tracemalloc/getsizeof 实测 __slots__ 与生成器的内存收益，用批量、异步与连接复用实测 I/O 提速，最后给出按瓶颈类型选手段的决策表。

Leeting Yan 7 分钟阅读 3141 字

本节目标：把剖析定位到的热点归类成 CPU、内存、I/O 三类，对每一类给出可量化的优化手法——向量化与算法改进、__slots__ 与生成器、批量与异步与连接复用，每条都附优化前后的真实数字。
适用版本：Python 3.12+（实测 3.14.6）；numpy 2.5.3、httpx 0.28.1

16.2 CPU·内存·I/O 三类瓶颈的定位与优化

16.1 告诉你「哪里慢」，但「怎么改」取决于瓶颈的性质。同一个热点函数，CPU 密集和 I/O 密集的解法截然相反：前者要靠算法和向量化，后者要靠并发和批量。这一节先把瓶颈分类，再对每一类给出经过实测的优化手段。

16.2.1 先分类，再动手

判断瓶颈类型最快的办法是看它把时间花在哪：

类型现象判断方法优化方向
CPU 密集单核跑满、tottime 高profiler 显示大量计算函数向量化、算法、多进程
内存密集RSS 持续上涨、频繁 GCtracemalloc 峰值高减少对象、__slots__、流式处理
I/O 密集CPU 空闲、等待占大头时间线里大量 sleep/read并发、批量、连接复用

误判的代价很大：给 I/O 密集的任务套上多进程（CPU 密集解法），只会徒增进程开销；给 CPU 密集的任务堆线程（I/O 密集解法），会被 GIL 卡死。先分类，再选手段。

16.2.2 CPU 密集：向量化与算法改进

CPU 瓶颈的第一杠杆是把 Python 层的循环下沉到 C 层。看一个逐元素计算的例子，三种写法对比：

import time
import numpy as np

N = 1_000_000

def pure_python(a, b):
    out = []
    for i in range(len(a)):
        out.append((a[i] + b[i]) * 0.5)
    return out

def list_comprehension(a, b):
    return [(a[i] + b[i]) * 0.5 for i in range(len(a))]

def vectorized(a, b):
    return (a + b) * 0.5          # 整个运算在 C 层完成

def timeit(fn, *args, repeat=3):
    best = float("inf")
    for _ in range(repeat):
        t0 = time.perf_counter()
        fn(*args)
        best = min(best, time.perf_counter() - t0)
    return best

真跑（本机实测，100 万元素）：

纯 for 循环     :     55.4 ms
列表推导式      :     52.4 ms  (1.1x)
numpy 向量化    :      0.4 ms  (123x)

三条结论：列表推导式比显式 append 快，但只快 10% 量级（它仍是一次次调用 Python 字节码）；真正的数量级跃迁来自 numpy 向量化，123 倍——因为 (a+b)*0.5 整个在 C 层对连续内存做 SIMD 运算，不经过 Python 解释器。「用推导式代替循环」是 10% 的优化，「换成向量化」是 100 倍的优化，优先级完全不同。

第二个杠杆是算法复杂度。同一个「判重」需求，两种实现：

def has_dup_quadratic(xs):          # O(n^2)
    for i in range(len(xs)):
        for j in range(i + 1, len(xs)):
            if xs[i] == xs[j]:
                return True
    return False

def has_dup_linear(xs):             # O(n)
    seen = set()
    for x in xs:
        if x in seen:
            return True
        seen.add(x)
    return False

真跑（本机实测，输入保证全部唯一，必须扫完）：

n= 1000  O(n^2)=  10.14 ms   O(n)= 0.044 ms   加速  230.0x
n= 2000  O(n^2)=  41.27 ms   O(n)= 0.084 ms   加速  488.7x
n= 3000  O(n^2)=  93.16 ms   O(n)= 0.129 ms   加速  721.9x
n=60000  O(n)  set 判重 = 1.865 ms

看 O(n^2) 那一列：n 翻倍，耗时约翻 4 倍（10→41→93，接近 4 倍与 9 倍），这就是平方级的行为；O(n) 那一列则近似线性。加速比随规模增长（230x → 489x → 722x）——这揭示了一条铁律：算法改进的收益随数据规模放大，而代码层面的微优化收益与规模无关。数据量越大，越该优先动算法。

16.2.3 内存：实测对象开销

内存瓶颈常被忽视，直到 OOM 打挂服务。先用 sys.getsizeof 和 tracemalloc 把「一个对象到底占多少」测出来——注意 getsizeof 只算对象本身，不含它引用的东西：

import sys, tracemalloc

class RowDict:
    def __init__(self, uid, name, score):
        self.uid, self.name, self.score = uid, name, score

class RowSlots:
    __slots__ = ("uid", "name", "score")
    def __init__(self, uid, name, score):
        self.uid, self.name, self.score = uid, name, score

d = RowDict(1, "alice", 95.5)
s = RowSlots(1, "alice", 95.5)
print(f"普通实例 sizeof       : {sys.getsizeof(d)} B  (+ __dict__ {sys.getsizeof(d.__dict__)} B)")
print(f"__slots__ 实例 sizeof : {sys.getsizeof(s)} B  (无 __dict__)")

真跑（本机实测）：

普通实例 sizeof       : 48 B  (+ __dict__ 296 B)
__slots__ 实例 sizeof : 56 B  (无 __dict__)

这里有个反直觉的坑：单看实例本身，__slots__ 版本反而「更大」（56 > 48）。因为普通实例把属性存在外挂的 __dict__ 里，实例本体很小；__slots__ 把三个槽位内联进对象，本体变大，但省掉了整个 296 字节的 __dict__。所以真正的对比要看总量——用 tracemalloc 建 10 万个实例：

def build(cls, n):
    return [cls(i, "x", i * 1.0) for i in range(n)]

for cls, label in ((RowDict, "dict"), (RowSlots, "slots")):
    tracemalloc.start()
    objs = build(cls, 100_000)
    cur, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    print(f"{label:6s} 10 万实例 tracemalloc 峰值: {peak / 1024 / 1024:6.2f} MB")
    del objs

真跑（本机实测）：

dict   10 万实例 tracemalloc 峰值:  15.25 MB
slots  10 万实例 tracemalloc 峰值:  11.43 MB

10 万个实例省下约 3.8 MB，约 25%——这个数字比「节省一半」的宣传保守得多，但它是实测的。__slots__ 的收益取决于属性个数：属性越多、实例越多，省得越多；只有两三个属性时收益有限。别信宣传，测你的场景。

第二个内存杠杆是用生成器替代列表，把「一次性物化」变成「流式消费」：

import sys, tracemalloc
N = 1_000_000

tracemalloc.start()
lst = [i * i for i in range(N)]        # 一次性生成 100 万个 int
cur, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
print(f"list 峰值: {peak / 1024 / 1024:.1f} MB")
del lst

tracemalloc.start()
total = sum(i * i for i in range(N))   # 生成器，边算边丢
cur, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
print(f"generator 峰值: {peak / 1024 / 1024:.3f} MB, 结果={total}")

真跑（本机实测）：

list 峰值: 38.6 MB
generator 峰值: 0.000 MB, 结果=333332833333500000

生成器峰值几乎为 0——它不物化任何中间列表，每算一个就交给 sum 丢掉。凡是「边生成边消费、不需要回头访问」的场景，都该用生成器或 yield：读大文件逐行处理、管道式的数据转换、流式聚合，都属于这一类。

16.2.4 I/O 密集：批量、异步、连接复用

I/O 瓶颈的特征是CPU 闲着在等。最常见的浪费是串行地发请求：每个请求要等上一个回来才发下一个。用本地 http.server 起一个「每次处理 20ms」的服务来实测三种模式：

import asyncio, threading, time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import httpx

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        time.sleep(0.02)                 # 模拟 20ms 后端处理
        body = b'{"ok": true}'
        self.send_response(200)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
    def log_message(self, *args): pass

def serve():
    srv = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv

三种客户端模式（串行复用连接、串行每次新建连接、异步并发）：

def sequential(url, n=50):               # 复用一条长连接
    with httpx.Client() as c:
        for _ in range(n):
            c.get(url)

def sequential_no_reuse(url, n=50):      # 每次请求新建连接
    for _ in range(n):
        with httpx.Client() as c:
            c.get(url)

async def concurrent_async(url, n=50):   # 50 个请求并发
    async with httpx.AsyncClient() as c:
        await asyncio.gather(*[c.get(url) for _ in range(n)])

真跑（本机实测，50 个请求）：

串行 + 连接复用     :  2018.1 ms
串行 + 每次新连接   :  4529.8 ms
异步并发 + 连接复用 :   216.9 ms
异步相对串行加速    :   9.3x

三个数字对应三条独立的优化：

  1. 连接复用（2018ms vs 4530ms，省 55%）：每次新建 TCP 连接要做三次握手，httpx.Client() 复用连接池就跳过了这部分。客户端要复用、服务端要开 keep-alive。
  2. 异步并发（2018ms → 217ms，9.3x）：50 个「等 20ms」的请求并发发出，总耗时从「50 × 20ms」压到「接近 1 × 20ms」。这是 I/O 瓶颈的标准解法——不是让单个请求更快，而是让等待重叠。
  3. 批量（未在上表，但同理）：能一次查 100 条就别查 100 次。数据库的 IN 查询、缓存的 mget、对象存储的批量接口，都是把 N 次往返压成 1 次。

提醒：httpx.Client 与 AsyncClient 都应作为长生命周期对象复用（模块级或依赖注入），别在循环里反复 with。用完后 await client.aclose() 关闭。

16.2.5 按瓶颈类型选手段

把上面的结论收成一张决策表，遇到性能问题照着选：

瓶颈首选手段实测收益量级次选
CPU · 逐元素计算numpy 向量化100x+array/memoryview
CPU · 重复查找set/dict 替代嵌套循环随规模增长（200x+）排序 + 二分
CPU · 纯计算多进程 ProcessPoolExecutor接近核数C 扩展 / Cython
内存 · 大量小对象__slots__~25%（本机实测）namedtuple
内存 · 大中间结果生成器 / 流式处理峰值趋近 0分块处理
I/O · 多次往返异步并发 / 批量5–10x线程池
I/O · 连接开销连接池复用~2xHTTP/2 多路复用

一条贯穿全表的原则：先分类，再选手段；先用算法/架构层面的优化，再抠代码细节——因为前者的收益是数量级，后者是百分比。

延伸阅读

小结

  • 先分类再动手：CPU 密集、内存密集、I/O 密集的解法互不通用，误判方向会白费力气。
  • CPU 的两级杠杆：代码层微优化（推导式）约 10%，算法/向量化是 100 倍；收益随数据规模放大。
  • numpy 向量化 123x（本机实测）：把逐元素循环下沉到 C 层；算法改进在判重场景达 230–722x。
  • 内存要看总量不看单例：__slots__ 单例 sizeof 反而更大，但 10 万实例省约 25%；生成器把 38.6MB 峰值压到接近 0。
  • I/O 靠重叠不靠提速：连接复用省 55%，异步并发在 50 请求上达 9.3x，批量把 N 次往返压成 1 次。
  • 优化顺序：算法/架构 > 数据结构 > 并发模型 > 代码细节，收益从数量级递减到百分比。

到这里我们知道「怎么让单次请求更快」。但真实系统的问题往往不是「单个请求慢」，而是「并发一上来就雪崩」——下一节用自写压测器把服务的容量曲线打出来，并用限流与降级守住它。

阅读导航：上一节：剖析方法论 · 下一节：压测、容量评估与限流降级 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. 《Python高级编程》目录
  2. 《Python高级编程》11.3 PEP 流程与版本迁移策略
  3. 《Python高级编程》11.2 嵌入式与自由线程运行时