【Python】第 3 章：内存管理
·
第 3 章：内存管理
3.1 引用计数
原理讲解
引用计数是 CPython 的主要内存管理机制。
┌─────────────────────────────────────────────────────────┐
│ 引用计数工作原理 │
├─────────────────────────────────────────────────────────┤
│ │
│ 对象 A (refcount=3) │
│ ↑ │
│ │ │
│ ┌───┼───┬───────────┐ │
│ │ │ │ │ │
│ x y z │ │
│ │ │
│ 当 refcount → 0: │
│ 1. 调用 tp_dealloc │
│ 2. 释放内存 │
│ 3. 递归减少引用对象的计数 │
│ │
└─────────────────────────────────────────────────────────┘
引用计数增减时机：
# 引用计数 +1 的情况：
a = obj # 赋值
func(obj) # 函数参数
container.append(obj) # 放入容器
return obj # 返回值
# 引用计数 -1 的情况：
del a # 删除变量
a = something_else # 重新赋值
函数返回 # 参数出栈
容器删除元素 # 从容器移除
引用计数的 C 实现：
// 增加引用计数
#define Py_INCREF(op) ( \
_Py_INC_REFTOTAL _Py_COUNT_ALLOCS_COMMA \
_Py_INC_OP_REFCNT(op))
// 减少引用计数
#define Py_DECREF(op) \
do { \
if (_Py_DEC_REFTOTAL _Py_COUNT_ALLOCS_COMMA \
_Py_DEC_OP_REFCNT(op) == 0) \
_Py_Dealloc(op); \
} while (0)
引用计数的问题 (循环引用)
# 循环引用示例
class Node:
def __init__(self, value):
self.value = value
self.ref = None
# 创建循环引用
a = Node("A")
b = Node("B")
a.ref = b # a → b
b.ref = a # b → a
# 删除外部引用
del a, b
# 问题：两个对象的引用计数都是 1（互相引用）
# 永远不会被引用计数机制回收！
循环引用图示：
删除外部引用后：
┌─────────┐ ┌─────────┐
│ Node A │ ──────→ │ Node B │
│ refcnt=1│ │ refcnt=1│
└────┬────┘ └────┬────┘
│ │
└───────────────────┘
互相引用，refcount 永不为 0
需要垃圾回收器处理
3.2 垃圾回收 (GC)
标记 - 清除算法
算法步骤：
┌─────────────────────────────────────────────────────────┐
│ 标记 - 清除算法流程 │
├─────────────────────────────────────────────────────────┤
│ │
│ 1. 标记阶段 (Mark) │
│ ┌─────────────────────────────────────────┐ │
│ │ 从根对象 (globals, stack, registers) │ │
│ │ 开始遍历，标记所有可达对象 │ │
│ │ │ │
│ │ 根对象 → A → B → C │ │
│ │ ↓ │ │
│ │ D (被标记) │ │
│ └─────────────────────────────────────────┘ │
│ │
│ 2. 清除阶段 (Sweep) │
│ ┌─────────────────────────────────────────┐ │
│ │ 遍历所有对象： │ │
│ │ - 已标记 → 清除标记，保留 │ │
│ │ - 未标记 → 释放内存 │ │
│ │ │ │
│ │ E (未标记) → 释放! │ │
│ │ F (未标记) → 释放! │ │
│ └─────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
Python 的 GC 实现：
import gc
# 查看 GC 统计
print(gc.get_stats())
# 手动触发 GC
gc.collect()
# 查看不可达对象
unreachable = gc.collect()
print(f"回收了 {unreachable} 个对象")
分代回收
三代回收机制：
┌─────────────────────────────────────────────────────────┐
│ 分代回收策略 │
├─────────────────────────────────────────────────────────┤
│ │
│ Generation 0 (新生代) │
│ ├── 新创建的对象 │
│ ├── 频繁回收 │
│ └── 阈值：700 次分配 │
│ │
│ Generation 1 (中生代) │
│ ├── 从 Gen0 存活的对象 │
│ ├── 较少回收 │
│ └── 阈值：10 次 Gen0 回收 │
│ │
│ Generation 2 (老年代) │
│ ├── 从 Gen1 存活的对象 │
│ ├── 很少回收 │
│ └── 阈值：10 次 Gen1 回收 │
│ │
│ 假设： │
│ - 大多数对象生命周期短（朝生夕死） │
│ - 存活越久的对象越可能继续存活 │
│ │
└─────────────────────────────────────────────────────────┘
import gc
# 查看各代阈值
print(gc.get_threshold()) # (700, 10, 10)
# 查看各代对象数量
for i, gen in enumerate(gc.get_count()):
print(f"Generation {i}: {gen} objects")
# 调整阈值
gc.set_threshold(1000, 15, 15)
gc 模块使用
import gc
# 1. 控制 GC 开关
gc.enable() # 启用
gc.disable() # 禁用
gc.isenabled() # 检查状态
# 2. 手动回收
collected = gc.collect() # 返回回收的对象数
print(f"回收了 {collected} 个对象")
# 3. 调试 GC
gc.set_debug(gc.DEBUG_STATS) # 打印统计信息
gc.set_debug(gc.DEBUG_LEAK) # 检测内存泄漏
# 4. 查看 GC 对象
print(gc.get_objects()) # 所有被 GC 跟踪的对象
print(gc.get_referrers(obj)) # 引用 obj 的对象
print(gc.get_referents(obj)) # obj引用的对象
# 5. 排除特定对象
class IgnoredClass:
pass
gc.collect()
ignored = IgnoredClass()
gc.collect([ignored]) # 不回收 ignored
3.3 内存池
小对象分配 (PyMalloc)
内存池架构：
┌─────────────────────────────────────────────────────────┐
│ CPython 内存池架构 │
├─────────────────────────────────────────────────────────┤
│ │
│ 用户请求 │
│ ↓ │
│ ┌─────────────────────────────────────────┐ │
│ │ PyObject_Malloc │ │
│ └─────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────┐ │
│ │ 对象大小 <= 512 bytes? │ │
│ └─────────────────────────────────────────┘ │
│ │ │ │
│ │ YES │ NO │
│ ↓ ↓ │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ 内存池 │ │ 系统 malloc │ │
│ │ (Arena) │ │ (大块分配) │ │
│ │ │ └─────────────┘ │
│ │ ┌───────┐ │ │
│ │ │ Pool │ │ 内存池分层： │
│ │ │ ├────┤ │ - Arena (4MB) │
│ │ │ │Block│ │ - Pool (4KB) │
│ │ │ ├────┤ │ - Block (8-512 bytes) │
│ │ │ │Block│ │ │
│ │ │ └────┘ │ │
│ │ └───────┘ │ │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
内存池优势：
- 减少碎片：固定大小的块分配
- 快速分配：从空闲链表获取，无需系统调用
- 批量管理：一次性释放整个 Pool
内存碎片问题
碎片产生：
初始：[AAAA][BBBB][CCCC][DDDD]
删除 B 和 D：[AAAA][ ][CCCC][ ]
尝试分配 E(大)：无法放入空闲块！
Python 的解决方案：
- 小对象使用内存池（固定大小块）
- 大对象直接使用系统 malloc
- 定期 GC 整理内存
内存优化技巧
# 技巧 1: 使用 __slots__
class WithoutSlots:
def __init__(self):
self.x = 1
self.y = 2
class WithSlots:
__slots__ = ['x', 'y']
def __init__(self):
self.x = 1
self.y = 2
# 技巧 2: 使用生成器代替列表
def create_list():
return [x * 2 for x in range(1000000)]
def create_generator():
return (x * 2 for x in range(1000000))
# 技巧 3: 及时删除大对象
def process_data():
data = load_huge_data()
result = process(data)
del data # 提前释放
return result
# 技巧 4: 使用 array 代替 list 存储数字
import array
numbers_list = [1, 2, 3, 4, 5] # 每个是 PyObject
numbers_array = array.array('i', [1, 2, 3, 4, 5]) # 紧凑存储
3.4 实践：使用 tracemalloc 分析内存使用
实验代码
# examples/chapter-03/memory_analysis.py
import tracemalloc
import gc
def analyze_memory():
"""使用 tracemalloc 分析内存使用"""
# 开始追踪
tracemalloc.start()
# 获取初始快照
snapshot1 = tracemalloc.take_snapshot()
# 创建一些对象
data = []
for i in range(10000):
data.append({'id': i, 'value': f'item_{i}' * 10})
# 获取当前快照
snapshot2 = tracemalloc.take_snapshot()
# 比较差异
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
print("内存使用 TOP 10:")
for stat in top_stats[:10]:
print(stat)
# 详细统计
print("\n总体统计:")
current, peak = tracemalloc.get_traced_memory()
print(f"当前内存：{current / 1024 / 1024:.2f} MB")
print(f"峰值内存：{peak / 1024 / 1024:.2f} MB")
# 停止追踪
tracemalloc.stop()
def compare_structures():
"""比较不同数据结构的内存使用"""
tracemalloc.start()
# 列表
list_data = list(range(10000))
snapshot_list = tracemalloc.take_snapshot()
list_mem, _ = tracemalloc.get_traced_memory()
# 元组
tuple_data = tuple(range(10000))
snapshot_tuple = tracemalloc.take_snapshot()
tuple_mem, _ = tracemalloc.get_traced_memory()
# 集合
set_data = set(range(10000))
snapshot_set = tracemalloc.take_snapshot()
set_mem, _ = tracemalloc.get_traced_memory()
print(f"列表内存：{list_mem / 1024:.2f} KB")
print(f"元组内存：{tuple_mem / 1024:.2f} KB")
print(f"集合内存：{set_mem / 1024:.2f} KB")
tracemalloc.stop()
def find_memory_leak():
"""模拟和检测内存泄漏"""
tracemalloc.start()
# 模拟泄漏
leaked_objects = []
for i in range(100):
# 创建对象但不清理
leaked_objects.append([0] * 10000)
if i % 10 == 0:
current, peak = tracemalloc.get_traced_memory()
print(f"Iteration {i}: {current / 1024 / 1024:.2f} MB")
# 检查引用
print(f"\n泄漏对象数量：{len(leaked_objects)}")
tracemalloc.stop()
if __name__ == "__main__":
print("=== 内存分析 ===")
analyze_memory()
print("\n=== 结构比较 ===")
compare_structures()
print("\n=== 内存泄漏检测 ===")
find_memory_leak()
实验练习
练习 1：分析循环引用的内存
import tracemalloc
import gc
def test_circular_reference():
class Node:
def __init__(self, value):
self.value = value
self.ref = None
tracemalloc.start()
# 创建循环引用
nodes = []
for i in range(1000):
a = Node(i)
b = Node(i + 1)
a.ref = b
b.ref = a
nodes.append(a)
snapshot1 = tracemalloc.take_snapshot()
# 删除外部引用
del nodes
# 不运行 GC
snapshot2 = tracemalloc.take_snapshot()
# 运行 GC
gc.collect()
snapshot3 = tracemalloc.take_snapshot()
print("删除后（无 GC）:")
for stat in snapshot2.compare_to(snapshot1, 'lineno')[:5]:
print(stat)
print("\nGC 后:")
for stat in snapshot3.compare_to(snapshot1, 'lineno')[:5]:
print(stat)
tracemalloc.stop()
test_circular_reference()
练习 2：比较不同字符串创建的内存
import tracemalloc
def method1():
"""使用 + 拼接"""
s = ""
for i in range(1000):
s += str(i)
return s
def method2():
"""使用 join"""
return "".join(str(i) for i in range(1000))
def method3():
"""使用列表收集"""
parts = []
for i in range(1000):
parts.append(str(i))
return "".join(parts)
for i, method in enumerate([method1, method2, method3], 1):
tracemalloc.start()
method()
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
print(f"方法{i}: 当前={current/1024:.2f}KB, 峰值={peak/1024:.2f}KB")
练习 3：监控长时间运行的内存
import tracemalloc
import time
def long_running_task():
tracemalloc.start()
start_time = time.time()
iteration = 0
while time.time() - start_time < 5: # 运行 5 秒
# 模拟工作
data = [x ** 2 for x in range(1000)]
result = sum(data)
iteration += 1
if iteration % 100 == 0:
current, peak = tracemalloc.get_traced_memory()
print(f"Iter {iteration}: {current/1024/1024:.2f}MB / "
f"Peak: {peak/1024/1024:.2f}MB")
tracemalloc.stop()
print(f"总迭代次数：{iteration}")
long_running_task()
常见问题
Q1: 为什么禁用 GC 后程序更快？
A: GC 有开销，如果你的程序：
- 没有循环引用
- 对象生命周期清晰
可以禁用 GC 提升性能：
gc.disable()
# ... 运行代码 ...
gc.collect() # 最后清理一次
Q2: 如何检测内存泄漏？
A:
import tracemalloc
import gc
tracemalloc.start()
# 运行代码
# ...
# 强制 GC
gc.collect()
# 检查是否还有未释放的内存
current, peak = tracemalloc.get_traced_memory()
print(f"GC 后仍有 {current} bytes 未释放")
Q3: 为什么小整数不需要 GC？
A:
- 小整数（-5 到 256）被缓存
- 不会被销毁，直到解释器退出
- 不在 GC 跟踪范围内
Q4: 弱引用如何帮助内存管理？
A:
import weakref
class Cache:
def __init__(self):
self._cache = weakref.WeakValueDictionary()
def store(self, key, obj):
self._cache[key] = obj
# 当 obj 无其他引用时，自动从缓存移除
Q5: 分代回收的阈值如何调优？
A:
# 更频繁的回收（适合短生命周期对象多）
gc.set_threshold(400, 5, 5)
# 更少的回收（适合长运行服务）
gc.set_threshold(1000, 20, 20)
# 完全禁用自动回收
gc.set_threshold(0, 0, 0)
本章小结
- 引用计数是 CPython 的主要内存管理机制
- 循环引用需要垃圾回收器处理
- 标记 - 清除算法识别不可达对象
- 分代回收基于"对象越老越可能存活"的假设
- 内存池优化小对象分配，减少碎片
- tracemalloc 是强大的内存分析工具
下一章预告
第 4 章将深入探讨 Python 数据结构实现，包括 list、dict、set 和 tuple 的底层原理。
这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。
更多推荐
14
0- 0
已为社区贡献3条内容
所有评论(0)