Skip to content

Python 内存管理 ​

1. 引用计数与 PyObject 结构 ​

Python 中一切皆对象,每个对象底层都是 PyObject 结构体:

c
// CPython 源码 Include/object.h
typedef struct _object {
    Py_ssize_t ob_refcnt;  // 引用计数
    PyTypeObject *ob_type; // 类型指针
} PyObject;

// 变长对象(如 list, str, tuple)
typedef struct {
    PyObject ob_base;
    Py_ssize_t ob_size;    // 元素个数(用于计算 len())
} PyVarObject;

引用计数增减规则:

python
import sys

a = []           # refcount = 1 (赋值)
b = a            # refcount = 2 (赋值)
print(sys.getrefcount(a))  # 3! 因为 getrefcount 参数本身也增加了一次

# 引用计数减少的情况:
# 1. del 语句
del b            # refcount = 1

# 2. 变量被重新赋值
c = [1, 2]
d = c            # refcount = 2
c = [3, 4]       # 原列表 refcount = 1

# 3. 离开作用域
def func():
    local = [1, 2]  # 函数返回后 refcount -> 0
func()

引用计数变为 0 时,对象立即被回收(deallocated)。


2. 循环引用 ​

引用计数的最大缺陷是无法处理循环引用:

python
class Node:
    def __init__(self):
        self.parent = None
        self.children = []

# 循环引用
parent = Node()
child = Node()
parent.children.append(child)
child.parent = parent

# del parent  # refcount(child) = 1 (parent.children 引用)
# del child   # refcount(parent) = 1 (child.parent 引用)
# 两者都无法通过引用计数回收!

这就是为什么需要 垃圾回收器(GC) 来检测和回收循环引用。


3. weakref — 弱引用 ​

python
import weakref

class ExpensiveObject:
    def __init__(self, name):
        self.name = name
    def __repr__(self):
        return f'ExpensiveObject({self.name})'
    def __del__(self):
        print(f'{self.name} 被回收')

# 创建弱引用
obj = ExpensiveObject('data')
weak = weakref.ref(obj)

print(weak())       # ExpensiveObject(data), 通过 weak() 访问实际对象
print(weak() is obj) # True

del obj             # 触发 __del__,对象被回收
print(weak())       # None

# WeakValueDictionary — 值为弱引用的字典
cache = weakref.WeakValueDictionary()
obj = ExpensiveObject('cached')
cache['key'] = obj
print(len(cache))   # 1
del obj             # 对象被回收后自动从字典中移除
print(len(cache))   # 0

常见用途:

  • 缓存(避免强引用导致内存泄漏)
  • 监听器模式(观察者不阻止被观察者回收)
  • 父子关系中的反向引用

4. 三代分代 GC ​

CPython 的垃圾回收器采用分代回收策略:

python
import gc

# 查看三代阈值
print(gc.get_threshold())  # (700, 10, 10)
# generation 0: 每 700 次分配触发
# generation 1: 每 10 次 gen0 回收触发
# generation 2: 每 10 次 gen1 回收触发

# 查看各代对象数量
print(gc.get_count())  # (当前各代对象计数)

# 手动触发回收
gc.collect()           # 完整回收所有代
gc.collect(0)          # 只回收第 0 代

# 查看 GC 统计
print(gc.get_stats())
# [{'collections': 0, 'collected': 0, 'uncollectable': 0}, ...]

分代假设:大部分对象生命周期短,存活越久的对象越不可能变成垃圾。

新对象 -> Generation 0
         (经过多次 GC 存活) -> Generation 1
                              (经过多次 GC 存活) -> Generation 2

禁用 GC(不推荐):

python
gc.disable()   # 禁用自动 GC
gc.enable()    # 重新启用
gc.isenabled() # 检查状态

5. GC Roots ​

垃圾回收器从 GC Roots 出发标记可达对象:

  1. 栈上的局部变量(当前函数调用栈)
  2. 全局变量(globals() 字典)
  3. 被 sys.getrefcount() 或 gc.get_referrers() 引用的对象
  4. C 扩展模块中的引用(解释器无法分析的 C 代码持有的引用)
python
import gc, sys

class Cyclic:
    def __init__(self):
        self.ref = None

# 创建循环引用但无 GC Root 指向
def make_cycle():
    a = Cyclic()
    b = Cyclic()
    a.ref = b
    b.ref = a
    # a, b 离开作用域,无 GC Root 指向

make_cycle()
gc.collect()  # GC 会回收这两个对象

6. 标记-清除算法 ​

CPython GC 使用标记-清除(Mark-Sweep)算法:

1. 暂停阶段:停止所有对象的引用变更
2. 标记阶段:从 GC Roots 遍历,标记所有可达对象
3. 清除阶段:遍历所有对象,回收未标记的对象
4. 恢复阶段:恢复引用变更
python
# 查找不可回收的对象
gc.set_debug(gc.DEBUG_LEAK)
gc.collect()

# 找出循环引用中不可回收的对象
unreachable = gc.garbage
print(f'不可回收对象数量: {len(unreachable)}')

注意事项:

  • GC 暂停会导致短暂的程序停顿
  • 大型对象图的 GC 暂停时间可能达到毫秒级
  • Python 3.4+ 已优化:不可回收的容器不会被收集到 gc.garbage

7. pymalloc — Arena / Pool / Block ​

CPython 使用 pymalloc 作为小对象的内存分配器(<= 512 字节):

Arena (256 KB)
└── Pool (4 KB, 一个内存页)
    └── Block (8/16/32/.../512 字节, 由 size class 决定)
python
import sys

# <= 512 字节使用 pymalloc
small = [1, 2, 3]
print(sys.getsizeof(small))  # 88, 使用 pymalloc

# > 512 字节直接使用系统 malloc
big = list(range(100))
print(sys.getsizeof(big))    # 856, 可能使用系统 malloc

# Arena 管理
# - 每个 Arena 是 256KB 的连续内存块
# - Arena 通过 freeArenas / usedArenas 链表管理
# - 当所有 Pool 都空闲时,Arena 会被返还给 OS

面试要点:pymalloc 通过按大小分组减少了内存碎片,但频繁创建/销毁大量不同大小的对象仍可能导致碎片化。


8. slots 原理 ​

python
# 不使用 __slots__ — 每个实例有 __dict__
class Regular:
    def __init__(self, x, y):
        self.x = x
        self.y = y

r = Regular(1, 2)
print(sys.getsizeof(r))       # 48 (只有 PyObject_HEAD)
print(sys.getsizeof(r.__dict__))  # 64 (额外的字典)

# 使用 __slots__ — 无 __dict__,属性存储在固定偏移
class Slotted:
    __slots__ = ('x', 'y')
    def __init__(self, x, y):
        self.x = x
        self.y = y

s = Slotted(1, 2)
print(sys.getsizeof(s))       # 40 (更小!)
# s.z = 3                     # AttributeError: 'Slotted' object has no attribute 'z'
特性有 __dict__有 __slots__
内存占用较大较小(节省 30-50%)
属性添加自由只能添加 __slots__ 中声明的
访问速度较慢(字典查找)较快(固定偏移)
继承兼容好需要手动声明父类 __slots__

9. 内存泄漏排查 — tracemalloc ​

python
import tracemalloc

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

# ... 业务代码 ...
data = [dict(zip(range(100), range(100))) for _ in range(10000)]

# 获取当前内存快照
snapshot = tracemalloc.take_snapshot()

# 按分配大小排序,显示 top 10
top_stats = snapshot.statistics('lineno')
print('--- Top 10 memory consumers ---')
for stat in top_stats[:10]:
    print(stat)

# 对比两个快照找增量
snapshot2 = tracemalloc.take_snapshot()
top_stats = snapshot.compare_to(snapshot2, 'lineno')
print('\n--- Top 10 memory increases ---')
for stat in top_stats[:10]:
    print(stat)

tracemalloc.stop()

其他工具:

  • objgraph:可视化对象引用关系
  • memory_profiler:逐行分析内存使用
  • gc.get_referrers():查找谁引用了某对象

10. Copy-on-Write (CoW) ​

Linux 的 fork() 使用 Copy-on-Write 机制,在多进程场景下影响内存使用:

python
import os
import sys

data = list(range(1000000))  # 大列表

pid = os.fork()
if pid == 0:
    # 子进程
    print(f'子进程 PID={os.getpid()}, data 地址={id(data)}')
    data.append(999)  # 触发 CoW,子进程获得独立副本
    print(f'修改后 data 地址={id(data)}')
    os._exit(0)
else:
    os.waitpid(pid, 0)
    print(f'父进程 data 地址={id(data)}')
    print(f'data[0]={data[0]}, data[-1]={data[-1]}')

Python 3.7+ 优化:multiprocessing 使用 os.fork() 时,子进程的内存页面在实际修改前不会复制。但某些操作(如引用计数更新)可能意外触发 CoW。

python
# 保持 CoW 友好:避免修改全局数据的引用计数
# 不好:
for item in global_list:  # 增加每个 item 的引用计数
    pass

# 更好:
for i in range(len(global_list)):
    item = global_list[i]  # 临时引用,不持久化