Python 内存管理、垃圾回收与性能调优:写出内存友好的 Python 代码

深入 Python 内存管理机制:引用计数、循环引用检测、分代垃圾回收、__slots__ 内存优化、对象池模式、内存泄漏排查工具(tracemalloc、objgraph),以及6个实用的性能调优技巧。

Python 的内存管理是自动的,但"自动"不等于"高效"。理解引用计数和垃圾回收机制,学会使用 __slots__ 和对象池,掌握内存泄漏排查工具——这是从"会写 Python"到"写好 Python"的分水岭。


目录

  1. Python 内存模型概览
  2. 引用计数:基础与陷阱
  3. 垃圾回收:循环引用与分代 GC
  4. slots:节省内存的利器
  5. 弱引用:打破循环引用
  6. 内存泄漏排查实战
  7. 性能调优 6 大技巧
  8. 对象池与缓存策略

1. Python 内存模型概览

Python 程序内存布局:
┌──────────────────┐
│   栈(Stack)     │ ← 函数调用、局部变量、引用
├──────────────────┤
│   堆(Heap)      │ ← 所有 Python 对象实际存储在这里
│ ┌──────────────┐ │
│ │ 小对象池      │ │ ← -5 ~ 256 的 int、短字符串等
│ ├──────────────┤ │
│ │ 分代 GC 区域  │ │ ← 第 0/1/2 代对象
│ ├──────────────┤ │
│ │ 大对象区域    │ │ ← 超过阈值直接分配
│ └──────────────┘ │
├──────────────────┤
│   代码/常量区     │ ← 字节码、字符串常量
└──────────────────┘

2. 引用计数:基础与陷阱

2.1 引用计数机制

Python 主要使用引用计数来管理内存:每个对象有一个 ob_refcnt 字段,记录有多少引用指向它。当计数归零,对象立即被销毁。

import sys

a = [1, 2, 3]
print(sys.getrefcount(a))   # 2(a + getrefcount 参数)

b = a
print(sys.getrefcount(a))   # 3(+ b)

del b
print(sys.getrefcount(a))   # 2

2.2 引用计数增加的场景

场景引用计数变化
赋值给变量+1
添加到列表/字典+1
作为函数参数传递+1(临时)
创建对象的引用+1

2.3 循环引用陷阱

# 循环引用:a 引用 b,b 引用 a
class Node:
    def __init__(self, name):
        self.name = name
        self.next = None
    def __del__(self):
        print(f"{self.name} 被销毁")

a = Node("A")
b = Node("B")
a.next = b
b.next = a

del a
del b
# 没有打印"被销毁"!因为循环引用导致引用计数永不为 0

3. 垃圾回收:循环引用与分代 GC

3.1 gc 模块

import gc

# 查看 GC 状态
print(gc.isenabled())       # True

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

# 查看无法回收的对象
unreachable = gc.collect()
print(f"本次回收了 {unreachable} 个不可达对象")

# 查看当前所有对象
print(f"当前对象数: {len(gc.get_objects())}")

# 查看循环引用导致的垃圾
for obj in gc.garbage:
    print(f"垃圾对象: {obj}")

3.2 分代垃圾回收

Python 使用三代分代回收:

代数存活检查频率说明
第 0 代最频繁新创建的对象
第 1 代中等经过一次 GC 后存活的对象
第 2 代最少长期存活的对象
# 调整阈值
gc.set_threshold(700, 10, 10)
# 第 0 代对象超过 700 个时触发
# 第 0 代收集 10 次,收集 1 次第 1 代
# 第 1 代收集 10 次,收集 1 次第 2 代

3.3 打破循环引用的方法

# 方法 1:手动设为 None
class Node:
    def __init__(self, name):
        self.name = name
        self.next = None

a = Node("A")
b = Node("B")
a.next = b
b.next = a

# 清理时手动断开
a.next = None
b.next = None
del a
del b

# 方法 2:使用 weakref(见第 5 节)

# 方法 3:使用上下文管理器
class ManagedNode:
    def __init__(self, name):
        self.name = name
        self._next = None
    
    @property
    def next(self):
        return self._next
    
    @next.setter
    def next(self, node):
        self._next = node
    
    def disconnect(self):
        self._next = None

4. slots:节省内存的利器

4.1 为什么需要 slots

默认的 Python 类使用 __dict__ 存储属性,每个实例都有一个字典——内存开销大。

class NormalUser:
    def __init__(self, name, age, email):
        self.name = name
        self.age = age
        self.email = email

class SlotUser:
    __slots__ = ['name', 'age', 'email']
    
    def __init__(self, name, age, email):
        self.name = name
        self.age = age
        self.email = email

# 内存对比
import sys

normal = NormalUser("Alice", 25, "alice@example.com")
slot = SlotUser("Alice", 25, "alice@example.com")

print(f"Normal: {sys.getsizeof(normal)} bytes")
print(f"Slot: {sys.getsizeof(slot)} bytes")
# Slot 版本可以节省 40%~50% 内存!

4.2 slots 的限制与注意事项

class User:
    __slots__ = ['name', 'age']
    
    def __init__(self, name, age):
        self.name = name
        self.age = age

user = User("Alice", 25)
# user.email = "alice@example.com"   # ❌ AttributeError!

# 无法使用 __dict__(除非显式添加)
# print(user.__dict__)   # ❌ AttributeError

# 子类需要重新定义 __slots__
class VipUser(User):
    __slots__ = ['level']   # 继承父类 slots,添加自己的
    
    def __init__(self, name, age, level):
        super().__init__(name, age)
        self.level = level

4.3 slots 适用场景

  • ✅ 需要创建大量实例的类(如数据行、节点)
  • ✅ 实例的属性固定不变
  • ❌ 属性需要动态添加的类
  • ❌ 需要使用 __dict__ 的类

5. 弱引用:打破循环引用

import weakref

class Node:
    def __init__(self, name):
        self.name = name
        self._next = None
    
    def __del__(self):
        print(f"{self.name} 被销毁")
    
    @property
    def next(self):
        return self._next() if self._next else None
    
    @next.setter
    def next(self, node):
        self._next = weakref.ref(node) if node else None

a = Node("A")
b = Node("B")
a.next = b
b.next = a

del a
del b
# 这次会打印两次"被销毁"!因为弱引用不增加引用计数

5.1 弱引用回调

def on_destroy(ref):
    print(f"对象被销毁了: {ref}")

obj = Node("Temp")
weak = weakref.ref(obj, on_destroy)

del obj   # → 对象被销毁了: <weakref at ...>

5.2 WeakKeyDictionary / WeakValueDictionary

# 当 key 或 value 不再被其他地方引用时,自动从字典中移除

weak_dict = weakref.WeakValueDictionary()

obj = Node("Test")
weak_dict["test"] = obj

print(len(weak_dict))    # 1
del obj
print(len(weak_dict))    # 0(自动移除)

6. 内存泄漏排查实战

6.1 tracemalloc

import tracemalloc

# 开始追踪
tracemalloc.start()

# ... 运行代码 ...

# 获取当前内存分配快照
current, peak = tracemalloc.get_traced_memory()
print(f"当前: {current / 1024 / 1024:.1f} MB, 峰值: {peak / 1024 / 1024:.1f} MB")

# 查看分配最多的代码
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

print("\n内存分配 Top 10:")
for stat in top_stats[:10]:
    print(f"{stat.size / 1024:.1f} KB: {stat.traceback.format()[-1]}")

tracemalloc.stop()

6.2 对比两个快照定位泄漏

import tracemalloc

tracemalloc.start()

# 第一次快照
snapshot1 = tracemalloc.take_snapshot()

# ... 执行可疑代码 ...
for _ in range(1000):
    data = [0] * 10000   # 模拟内存分配

# 第二次快照
snapshot2 = tracemalloc.take_snapshot()

# 对比
diff = snapshot2.compare_to(snapshot1, 'lineno')
print("分配增量 Top 10:")
for stat in diff[:10]:
    if stat.size_diff > 0:
        print(f"+{stat.size_diff / 1024:.1f} KB: {stat.traceback.format()[-1]}")

6.3 objgraph 可视化引用关系

# pip install objgraph
import objgraph

# 列出增长最多的类型
objgraph.show_most_common_types(limit=10)

# 找到泄漏对象的引用链
objgraph.show_backrefs(
    [some_leaky_object],
    max_depth=10,
    filename='leak_backrefs.png'
)

7. 性能调优 6 大技巧

技巧 1:使用局部变量

# ❌ 每次循环都访问全局/属性
for i in range(1000000):
    result = math.sqrt(i)

# ✅ 缓存到局部变量
from math import sqrt
for i in range(1000000):
    result = sqrt(i)

技巧 2:避免在循环中创建对象

# ❌
result = []
for i in range(100000):
    result.append(str(i))   # 每次创建新 str

# ✅ 使用生成器
result = (str(i) for i in range(100000))

技巧 3:列表推导式优于循环

# ❌
squares = []
for x in range(1000):
    squares.append(x ** 2)

# ✅
squares = [x ** 2 for x in range(1000)]   # C 层循环,更快

技巧 4:使用 in 而非多次比较

# ❌
if x == 1 or x == 2 or x == 3:
    pass

# ✅
if x in {1, 2, 3}:   # set 的 O(1) 查找
    pass

技巧 5:字符串拼接用 join

# ❌ O(n²)
result = ""
for s in strings:
    result += s

# ✅ O(n)
result = "".join(strings)

技巧 6:使用 __slots__array

# 大数据量时节省内存
from array import array

# ❌ Python int 列表(每个 int 28 bytes)
numbers = [0] * 1000000

# ✅ array(每个元素 4 bytes)
numbers = array('i', [0]) * 1000000

# ✅ numpy(更强大的数组)
import numpy as np
numbers = np.zeros(1000000, dtype=np.int32)

8. 对象池与缓存策略

8.1 简单对象池

class ObjectPool:
    """对象池:复用对象减少创建开销"""
    
    def __init__(self, factory, max_size=100):
        self.factory = factory
        self.max_size = max_size
        self._available = []
        self._in_use = set()
    
    def acquire(self):
        if self._available:
            obj = self._available.pop()
        else:
            obj = self.factory()
        self._in_use.add(id(obj))
        return obj
    
    def release(self, obj):
        obj_id = id(obj)
        if obj_id in self._in_use:
            self._in_use.remove(obj_id)
            if len(self._available) < self.max_size:
                self._available.append(obj)

# 使用
pool = ObjectPool(lambda: bytearray(1024))
buf = pool.acquire()
# ... 使用 buf ...
pool.release(buf)

8.2 functools.lru_cache

from functools import lru_cache

# 自动缓存最近使用的调用结果
@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# 对于 I/O 操作同样有效
@lru_cache(maxsize=32)
def read_config(key):
    # 从数据库或文件读取
    import time
    time.sleep(0.1)
    return f"config_value_for_{key}"

延伸阅读


Python 的内存管理是自动的,但高效使用内存需要人为干预。理解引用计数、善用 __slots__、会用 tracemalloc 排查泄漏,是写出生产级 Python 代码的必备技能。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. Python 高级异步编程:Trio 结构化并发与 AnyIO 兼容层
  2. Python 数据工程与 ETL 管道实战
  3. Python 元编程与动态特性深度解析