张会挽's Blog

Python 内存管理详解：垃圾回收与性能优化

Posted on In

Python 虽然自带垃圾回收机制，但并不意味着可以完全忽略内存管理。在实际项目中，内存泄漏、内存占用过高的问题并不少见。理解 Python 的内存管理机制，是写出高效 Python 代码的基础。

引用计数机制

Python 使用引用计数（Reference Counting）作为主要的垃圾回收手段。

工作原理

每个 Python 对象都有一个引用计数。当对象被引用时，计数加 1；引用被删除时，计数减 1。当计数归零时，对象立即被回收。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import sys

# 创建对象
a = [1, 2, 3]  # 引用计数 = 1
print(sys.getrefcount(a))  # 2（getrefcount 本身也产生一次临时引用）

# 增加引用
b = a          # 引用计数 = 2
c = a          # 引用计数 = 3

# 减少引用
del b          # 引用计数 = 2
c = None       # 引用计数 = 1
del c          # 引用计数 = 0，对象被回收

引用计数的优缺点

优点：

  • 简单高效，零延迟回收
  • 对象无引用时立即释放，内存立即可用

缺点：

  • 无法处理循环引用
  • 每个对象都需要额外存储引用计数
  • 线程安全操作有开销
1
2
3
4
5
6
7
8
9
10
# 循环引用示例
a = []
b = [a]
a.append(b)  # a 引用 b，b 引用 a，循环引用

# 即使 del a 和 del b，对象也不会被立即回收
del a
del b
# 此时 a 和 b 互相引用，但已经无法访问
# 引用计数都是 1，但这个引用来自彼此，不是有效引用

循环垃圾回收器

为了解决循环引用问题，Python 引入了循环垃圾回收器（Cyclic Garbage Collector）。

分代回收

Python 将内存中的对象分为三代：

代 特点 回收频率
0 代 新创建的对象 最高
1 代 经历一次回收仍存在的对象 中等
2 代 经历多次回收仍存在的对象 最低 
1
2
3
4
5
6
7
8
import gc

# 获取各代对象数量
print(gc.get_count())    # (total, 0_gen, 1_gen)
print(gc.get_stats())    # 各代详细统计

# 手动触发回收
gc.collect()

回收过程

当 0 代对象数量超过阈值时，触发垃圾回收：

  1. 暂停程序运行（Stop The World）
  2. 遍历所有对象，构建有向图
  3. 标记不可达的对象
  4. 回收不可达对象
  5. 升级幸存对象到下一代
1
2
3
4
5
import gc

# 设置各代回收阈值
gc.set_threshold(700, 10, 10)
# 0 代阈值 700，1 代阈值 10%（相对于 0 代），2 代阈值 10%（相对于 1 代）

分代回收的原理

1
2
3
4
5
6
7
8
# 对象的 header 结构（简化）
typedef struct {
    PyObject_HEAD        # 引用计数 + 类型指针
    PyGC_Head gc_head;   # 垃圾回收相关（prev, next 指针）
} PyObject;

# gc_head 构成双向链表，同一代对象链接在一起
# 0 代 -> 1 代 -> 2 代

常见内存泄漏场景

即使有垃圾回收器，以下场景仍会导致内存泄漏。

场景1：全局变量持有引用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# 错误写法
cached_data = []

def process(item):
    cached_data.append(item)  # 不断累积，永不清理
    return process_logic(item)

# 正确写法
cached_data = []

def process(item):
    cached_data.append(item)
    try:
        return process_logic(item)
    finally:
        cached_data.clear()  # 处理完清理

# 或者使用弱引用
import weakref

cached_data = weakref.WeakSet()

def process(item):
    cached_data.add(item)
    return process_logic(item)

场景2：类属性持有引用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 错误写法
class DataProcessor:
    results = []  # 类属性，所有实例共享

    def process(self, data):
        self.results.append(data)  # 不断累积

# 正确写法
class DataProcessor:
    def __init__(self):
        self.results = []  # 实例属性

    def process(self, data):
        self.results.append(data)

场景3：闭包捕获外部变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# 错误写法
def create_handler():
    data = []

    def handler(item):
        data.append(item)  # 闭包持有 data 引用
        return process(item)

    return handler

# handler 函数对象始终持有 data 列表的引用
# data 永远不会被回收

# 正确写法
def create_handler():
    data = []

    def handler(item):
        data.append(item)
        try:
            return process(item)
        finally:
            data.clear()  # 显式清理

# 或者使用弱引用
import weakref

def create_handler():
    data = weakref.ref(list())

    def handler(item):
        current = data()
        if current is None:
            return process(item)
        current.append(item)
        return process(item)

    return handler

场景4：监听器未注销

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# GUI/事件系统常见问题
class EventManager:
    def __init__(self):
        self.listeners = []

    def add_listener(self, callback):
        self.listeners.append(callback)

    def remove_listener(self, callback):
        self.listeners.remove(callback)  # 必须显式移除

# 问题
manager = EventManager()
obj = SomeObject()
manager.add_listener(obj.on_event)  # manager 持有 obj 的引用
del obj  # obj 不会被回收，因为还在 listeners 中

内存分析工具

tracemalloc

Python 3.4+ 内置的内存追踪工具：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import tracemalloc

# 启动追踪
tracemalloc.start()

# ... 执行代码 ...

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

# 打印前10大内存占用
top_stats = snapshot.statistics('lineno')
print("\nTop 10 内存占用:")
for stat in top_stats[:10]:
    print(stat)

# 查找特定文件的内存占用
for stat in snapshot.statistics('filename'):
    if 'mymodule' in str(stat):
        print(stat)

# 比较两个时间点的差异
snapshot1 = tracemalloc.take_snapshot()
# ... 执行代码 ...
snapshot2 = tracemalloc.take_snapshot()

top_diff = snapshot2.compare_to(snapshot1, 'lineno')
print("\n内存增长前10:")
for stat in top_diff[:10]:
    print(stat)

tracemalloc.stop()

objgraph

第三方工具，用于可视化对象引用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 安装：pip install objgraph

import objgraph

# 查找增长最多的对象类型
objgraph.show_most_common_types(limit=10)

# 统计特定类型对象数量
print(f"字典数量: {objgraph.count('dict')}")

# 查看对象的引用链
objgraph.show_refs([my_object], filename='refs.png')

# 查看是什么引用持有对象
objgraph.show_backrefs([my_object], filename='backrefs.png')

muppy

专门用于内存分析：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 安装：pip install muppy

from pympler import muppy, summary

# 内存快照摘要
all_objects = muppy.get_objects()
sums = summary.summarize(all_objects)
summary.print_(sums)

# 内存增长检测
from pympler import tracker

tr = tracker.SummaryTracker()
# ... 执行代码 ...
tr.print_diff()

实战：优化大数据处理内存占用

问题背景

处理大型 CSV 文件时，一次性读取可能导致内存溢出：

1
2
3
4
5
6
import pandas as pd

# 问题代码：一次性读取全部数据
def bad_approach(file_path):
    df = pd.read_csv(file_path)  # 假设文件 10GB，这里会 OOM
    return df.groupby('category').sum()

优化方案

1. 分块读取

1
2
3
4
5
6
7
8
9
10
11
12
import pandas as pd

def chunked_approach(file_path):
    results = {}
    chunk_size = 100_000  # 每次处理 10 万行

    for chunk in pd.read_csv(file_path, chunksize=chunk_size):
        grouped = chunk.groupby('category')['value'].sum()
        for cat, val in grouped.items():
            results[cat] = results.get(cat, 0) + val

    return results

2. 指定数据类型

1
2
3
4
5
6
7
8
9
10
def typed_approach(file_path):
    # 降低内存占用的关键：使用合适的数据类型
    dtype = {
        'id': 'int32',      # int64 -> int32，节省 50%
        'category': 'category',  # 重复字符串使用 category 类型
        'value': 'float32'  # float64 -> float32，节省 50%
    }

    df = pd.read_csv(file_path, dtype=dtype)
    return df.groupby('category')['value'].sum()

3. 只读取需要的列

1
2
3
4
5
6
7
def selective_columns(file_path):
    # 只需要两列，不读取其他列
    df = pd.read_csv(
        file_path,
        usecols=['category', 'value']
    )
    return df.groupby('category')['value'].sum()

4. 使用生成器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import csv

def generator_approach(file_path):
    """完全不用 pandas，按行处理"""
    results = {}

    with open(file_path, 'r') as f:
        reader = csv.DictReader(f)
        for row in reader:
            cat = row['category']
            val = float(row['value'])
            results[cat] = results.get(cat, 0) + val

    return results

内存优化效果对比

假设 CSV 文件 10GB，1 亿行数据：

方案 内存占用 处理时间
原始 pandas.read_csv ~12 GB 最快
分块读取 ~200 MB 较慢（多次 IO）
指定数据类型 ~6 GB 最快
生成器 ~50 MB 取决于磁盘速度

总结

理解 Python 内存管理的要点：

  1. 引用计数：主要回收机制，即时但无法处理循环引用
  2. 分代回收：处理循环引用，0/1/2 三代，频繁回收新对象
  3. 常见泄漏：全局变量、类属性、闭包、监听器未注销
  4. 分析工具：tracemalloc（内置）、objgraph、muppy
  5. 大数据处理：分块读取、合适的数据类型、按需加载

合理的内存管理能让 Python 程序在资源受限环境中稳定运行。