一、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}")