一、Python GC的三层架构

Python的垃圾回收机制由引用计数(保证实时性)+ 标记-清除(解决循环引用)+ 分代回收(优化扫描效率)三层组成。理解其协作方式,才能设计出内存高效的程序。

1.1 引用计数:实时回收的主力

import sys

# Python对象的核心回收机制:引用计数
# 每个对象维护一个refcount,引用为0时立即释放

class Node:
    def __init__(self, val):
        self.val = val

n = Node(42)       # refcount = 1
print(sys.getrefcount(n))  # 2(getrefcount本身也持有一个引用)

n2 = n             # refcount = 2
n3 = n             # refcount = 3
del n               # refcount = 2
del n2              # refcount = 1
del n3              # refcount = 0 → 对象被立即回收!

# ⚠️ 引用计数的致命弱点:循环引用无法回收
class Parent:
    def __init__(self):
        self.child = None

p = Parent()
c = Parent()
p.child = c       # p.refcount不变(c是引用但p不是)
c.child = p        # c.refcount不变
# 形成循环:p ↔ c
del p
del c
# refcount都是1(互相持有),但外部引用已无
# → 引用计数无法回收 → 等待标记-清除处理

# 内存观察:循环引用对象泄漏
import gc
gc.collect()  # 主动触发GC
print(gc.garbage)  # 列出无法回收的对象

二、分代回收机制

2.1 三代设计与回收频率

# Python将所有对象分为3代(Generation 0/1/2)
# 新创建的对象在Gen0,存活越久越"老"

# 回收策略:
# Gen0对象达到阈值时 → 触发Minor GC → 回收Gen0存活对象→ 幸存者晋升Gen1
# Gen1达到阈值时      → 触发Major GC → 回收Gen0+Gen1
# Gen2                  → Major GC时一同处理

# 查看各代信息
import gc
print(gc.get_threshold())   # (700, 10, 10) → Gen0阈值700
print(gc.get_count())      # 当前各代对象数量

# 设置阈值(一般不需要调整)
gc.set_threshold(500, 10, 10)

# ⚠️ 常见误区:大量短生命周期对象导致频繁GC
def process():
    data = [LargeObject() for _ in range(10000)]
    # data超出作用域后引用计数归零,立即回收
    # 但如果对象被加入全局列表,引用计数永不归零
    # 解决方案:用局部变量 + del

# ✅ 主动禁用GC的场景:
# ① 性能关键代码段(GC暂停影响延迟)
gc.disable()
try:
    for _ in range(10000):
        process_batch()
finally:
    gc.enable()

# ② 批量导入大量模块时
# 模块导入时频繁创建对象,GC干扰启动速度

2.2 内存泄漏的常见模式

# ① 全局列表无限增长(最常见)
cache = []  # 全局变量

def add_to_cache(item):
    cache.append(item)  # ⚠️ 永不清理
# 10000次调用后,cache有10000个引用

# ✅ 带容量限制的缓存
from collections import deque

class BoundedCache:
    def __init__(self, maxsize=1000):
        self.cache = deque(maxlen=maxsize)

    def add(self, item):
        self.cache.append(item)  # 超过maxsize自动淘汰旧元素

# ② 类属性持有大对象引用
class Service:
    def __init__(self):
        self.large_data = load_large_data()  # 永不释放

    def process(self):
        result = self.large_data.do_something()
        return result
# 如果Service是单例,large_data永驻内存

# ✅ 正确:用完释放
class Service:
    def __init__(self):
        self._large_data = None

    def _get_data(self):
        if self._large_data is None:
            self._large_data = load_large_data()
        return self._large_data

    def process(self):
        return self._get_data().do_something()

    def unload(self):
        self._large_data = None  # 手动释放
        gc.collect()

三、内存分析与优化工具

# ① tracemalloc:Python内置内存追踪
import tracemalloc

tracemalloc.start()

# 执行代码
process_data()

snapshot = tracemalloc.take_snapshot()
# 找出占用最大的前10行
top_stats = snapshot.statistics('lineno')[:10]
for stat in top_stats:
    print(stat)

# ② objgraph:对象数量分析
import objgraph

print(f"dict数量: {objgraph.count('dict')}")
print(f"list数量: {objgraph.count('list')}")

# 找出增长最多的对象类型
objgraph.get_leaking_objects()  # 返回无法回收的对象

# ③ memory_profiler:逐行内存分析
# pip install memory_profiler
# @profile装饰器放在需要分析的函数上

# @profile
def memory_heavy_function():
    x = [i * 2 for i in range(1000000)]
    return sum(x)

# 运行:$ python -m memory_profiler example.py
# 显示每行代码的内存增量

# ④ tracemalloc + diff对比
import tracemalloc

tracemalloc.start()
snapshot1 = tracemalloc.take_snapshot()

heavy_operation()

snapshot2 = tracemalloc.take_snapshot()
top_diff = snapshot2.compare_to(snapshot1, 'lineno')[:5]
for stat in top_diff:
    print(f"+{stat.size_diff/1024:.1f}KB: {stat}")