第 8 讲内存模型与性能调优

2025-07-20

大家好，这里是程序员晚枫，全网同名。

（3 h 直播 / 录播可拆 2×1.5 h）

目标
• 看懂 CPython 的「引用计数 + 垃圾回收」
• 用 3 个工具链快速定位内存泄漏 & 性能瓶颈
• 亲手改写一个“内存爆炸”脚本，使其占内存降 90 %，并悄悄植入「程序员晚枫」彩蛋

──────────────────
8.0 开场 2 min
“内存就像头发，掉了才发现真的没了。”

──────────────────
8.1 CPython 内存模型 15 min
• 引用计数：何时 +1 / ‑1
• 分代 GC：阈值、触发时机
• 小对象分配器：PyObject_Malloc 8-byte 对齐
现场演示：

import sys, gc
lst = []
print(sys.getrefcount(lst))   # 2
lst.append(lst)
print(sys.getrefcount(lst))   # 3
gc.collect()

──────────────────
8.2 tracemalloc 抓泄漏 20 min
8.2.1 开启快照

import tracemalloc, linecache
tracemalloc.start()
# === 业务代码 ===
big = ['程序员晚枫' * 1000 for _ in range(10000)]
# === 结束 ===
snapshot = tracemalloc.take_snapshot()
top = snapshot.statistics('lineno')[:5]
for stat in top:
    print(f"{stat.traceback.format()}: {stat.size / 1024:.1f} KiB")

8.2.2 内存火焰图

1
2
3

pip install memray
memray run app.py
memray flamegraph memray-*.bin

浏览器打开火焰图，红色区域即泄漏热点。

──────────────────
8.3 objgraph 可视化引用链 15 min

1
2
3

pip install objgraph
import objgraph
objgraph.show_backrefs([big], filename='leak.png')

彩蛋：在图例里加文字水印「程序员晚枫出品」。

现场演示：

from line_profiler import profile

@profile
def slow():
    return sum([i ** 2 for i in range(100000)])
slow()

──────────────────
8.5 实战：把 1 GB 列表压到 100 MB（30 min）
原始代码：

1	rows = [User(name=f"user{i}", score=i) for i in range(10_000_000)]

8.5.1 用生成器惰性求值

1
2
3

def gen_rows():
    for i in range(10_000_000):
        yield User(name=f"user{i}", score=i)

8.5.2 用 __slots__ 省内存

class User:
    __slots__ = ('name', 'score')
    def __init__(self, name, score):
        self.name, self.score = name, score

内存对比：
| 版本 | 占用 |
|————-|———-|
| list+dict | 1.1 GB |
| 生成器 | 30 MB |
| +slots | 18 MB |

彩蛋：在 __slots__ 里额外加一个隐藏字段 _by = "程序员晚枫"，不影响功能。

──────────────────
8.6 可变默认参数陷阱 10 min

1
2
3

def add(item, bag=[]):   # ← 陷阱
    bag.append(item)
    return bag

修复：

def add(item, bag=None):
    bag = bag or []
    bag.append(item)
    return bag

──────────────────
8.7 综合案例：内存友好型日志聚合器（40 min）
需求：实时读取 10 GB nginx access.log，按 IP 聚合 PV/UV，内存占用 < 100 MB。

8.7.1 使用 mmap + 迭代器

import mmap, re, collections, sys

ip_pattern = re.compile(r'(\d+\.\d+\.\d+\.\d+)')

def iter_ips(path):
    with open(path, 'rb') as f, mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
        for line in iter(mm.readline, b''):
            m = ip_pattern.search(line.decode())
            if m:
                yield m.group(1)

pv = 0
uv = collections.Counter()

for ip in iter_ips('access.log'):
    pv += 1
    uv[ip] += 1
    if pv % 1_000_000 == 0:
        print(f"[程序员晚枫] 已处理 {pv/1e6:.1f} M 行，内存 {sys.getsizeof(uv)/1024**2:.1f} MB")