🗑️

内存管理与垃圾回收

📂 python ⏱ 3 min 579 words

内存管理与垃圾回收

Python的内存管理是自动的，但理解其机制能帮助你写出更高效、更少内存泄漏的代码。

引用计数

Python主要使用引用计数来管理内存。每个对象都有一个引用计数器，记录有多少引用指向它。

import sys

class MyObject:
    def __init__(self, name):
        self.name = name
    
    def __del__(self):
        print(f"销毁对象: {self.name}")

# 创建对象
obj = MyObject("test")
print(sys.getrefcount(obj))  # 2（obj + getrefcount的参数）

# 增加引用
a = obj
print(sys.getrefcount(obj))  # 3

b = [obj, obj]
print(sys.getrefcount(obj))  # 4

# 删除引用
del a
print(sys.getrefcount(obj))  # 3

# 函数参数也会增加引用
def func(x):
    print(sys.getrefcount(x))  # 2（x + getrefcount的参数）

func(obj)
print(sys.getrefcount(obj))  # 3

引用计数的限制

import sys
import ctypes

# 查看引用计数
a = [1, 2, 3]
print(sys.getrefcount(a))  # 2

# 强制增加引用计数（危险操作）
ctypes.c_long.from_address(id(a)).value += 1
print(sys.getrefcount(a))  # 3

# 强制减少引用计数（非常危险）
ctypes.c_long.from_address(id(a)).value -= 1
print(sys.getrefcount(a))  # 2

# 注意：这种操作可能导致程序崩溃，仅用于演示

循环引用与垃圾回收

引用计数无法处理循环引用，因此Python还有分代垃圾回收器。

import gc

class Node:
    def __init__(self, name):
        self.name = name
        self.parent = None
        self.children = []
    
    def add_child(self, child):
        self.children.append(child)
        child.parent = self
    
    def __repr__(self):
        return f"Node({self.name})"

# 创建循环引用
parent = Node("parent")
child = Node("child")
parent.add_child(child)

# 删除外部引用
del parent
del child

# 垃圾回收器可能还没运行
print("删除引用后，对象仍然存在")

# 手动触发垃圾回收
gc.collect()
print(f"回收了 {gc.collect()} 个对象")

GC分代机制

import gc

# 查看GC统计信息
print("GC统计信息:")
print(f"  代0: {gc.get_count()[0]}")
print(f"  代1: {gc.get_count()[1]}")
print(f"  代2: {gc.get_count()[2]}")

# GC阈值
print(f"\nGC阈值: {gc.get_threshold()}")
# 默认为 (700, 10, 10)

# 手动控制GC
gc.disable()  # 禁用自动GC
gc.enable()   # 启用自动GC
gc.collect()  # 手动触发

# 设置阈值
gc.set_threshold(1000, 15, 15)  # 调整为更高阈值

弱引用

弱引用允许你引用对象而不增加其引用计数，常用于缓存。

import weakref

class ExpensiveObject:
    def __init__(self, name):
        self.name = name
        print(f"创建昂贵对象: {name}")
    
    def __del__(self):
        print(f"销毁对象: {self.name}")

# 创建弱引用
obj = ExpensiveObject("test")
weak = weakref.ref(obj)

print(f"对象存在: {weak() is not None}")  # True
print(f"弱引用对象: {weak()}")

# 删除强引用
del obj
print(f"对象存在: {weak() is not None}")  # False

弱引用字典

import weakref

class Data:
    def __init__(self, value):
        self.value = value
    
    def __repr__(self):
        return f"Data({self.value})"

# 使用WeakValueDictionary
cache = weakref.WeakValueDictionary()

# 创建对象并缓存
obj1 = Data(1)
obj2 = Data(2)

cache["key1"] = obj1
cache["key2"] = obj2

print(f"缓存大小: {len(cache)}")  # 2

# 删除一个对象
del obj1
print(f"删除obj1后缓存大小: {len(cache)}")  # 1
print(f"key1还在缓存中: {'key1' in cache}")  # False

弱引用回调

import weakref

class TrackedObject:
    def __init__(self, name):
        self.name = name

def callback(ref, name=""):
    print(f"对象 {name} 已被销毁")

obj = TrackedObject("test")
weak = weakref.ref(obj, lambda ref: callback(ref, "test"))

del obj  # 输出: 对象 test 已被销毁

内存泄漏检测

import gc

class LeakyClass:
    """有内存泄漏的类"""
    _instances = []
    
    def __init__(self, name):
        self.name = name
        LeakyClass._instances.append(self)  # 泄漏：静态列表持有引用

class FixedClass:
    """修复后的类"""
    _instances = weakref.WeakSet()
    
    def __init__(self, name):
        self.name = name
        FixedClass._instances.add(self)  # 使用弱引用

# 演示内存泄漏
for i in range(1000):
    obj = LeakyClass(f"leaky_{i}")

print(f"LeakyClass实例数: {len(LeakyClass._instances)}")  # 1000

# 修复后
for i in range(1000):
    obj = FixedClass(f"fixed_{i}")

gc.collect()
print(f"FixedClass实例数: {len(FixedClass._instances)}")  # 远小于1000

tracemalloc内存分析

import tracemalloc

# 开始内存跟踪
tracemalloc.start()

# 分配一些内存
data = [i ** 2 for i in range(10000)]
text = "Hello " * 1000

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

# 按内存使用量排序
top_stats = snapshot.statistics('lineno')

print("Top 10 内存使用:")
for stat in top_stats[:10]:
    print(stat)

详细内存分析

import tracemalloc

def process_data():
    """模拟数据处理"""
    large_list = [i for i in range(100000)]
    large_dict = {i: i*2 for i in range(50000)}
    return sum(large_list)

tracemalloc.start()

# 记录开始时的内存
snapshot1 = tracemalloc.take_snapshot()

# 执行操作
result = process_data()

# 记录结束时的内存
snapshot2 = tracemalloc.take_snapshot()

# 比较两个快照
stats = snapshot2.compare_to(snapshot1, 'lineno')

print("内存增长 Top 10:")
for stat in stats[:10]:
    print(stat)

tracemalloc.stop()

内存优化技巧

使用slots

import sys

class RegularClass:
    def __init__(self, x, y):
        self.x = x
        self.y = y

class SlottedClass:
    __slots__ = ['x', 'y']
    
    def __init__(self, x, y):
        self.x = x
        self.y = y

regular = RegularClass(1, 2)
slotted = SlottedClass(1, 2)

print(f"RegularClass大小: {sys.getsizeof(regular) + sys.getsizeof(regular.__dict__)} 字节")
print(f"SlottedClass大小: {sys.getsizeof(slotted)} 字节")

# 创建大量实例
import time

start = time.time()
regular_instances = [RegularClass(i, i*2) for i in range(100000)]
regular_time = time.time() - start

start = time.time()
slotted_instances = [SlottedClass(i, i*2) for i in range(100000)]
slotted_time = time.time() - start

print(f"RegularClass创建时间: {regular_time:.3f}秒")
print(f"SlottedClass创建时间: {slotted_time:.3f}秒")

使用生成器减少内存

import sys

# 列表推导式 - 一次性占用内存
list_comp = [i ** 2 for i in range(1000000)]
print(f"列表推导式内存: {sys.getsizeof(list_comp)} 字节")

# 生成器表达式 - 惰性求值
gen_exp = (i ** 2 for i in range(1000000))
print(f"生成器表达式内存: {sys.getsizeof(gen_exp)} 字节")

# 使用生成器函数
def squares(n):
    for i in range(n):
        yield i ** 2

gen_func = squares(1000000)
print(f"生成器函数内存: {sys.getsizeof(gen_func)} 字节")

内存管理最佳实践

及时删除不需要的引用：使用del或重新赋值
使用弱引用进行缓存：weakref.WeakValueDictionary
使用__slots__：减少实例内存占用
使用生成器：处理大数据集时避免内存溢出
定期调用gc.collect()：处理循环引用
使用tracemalloc：定位内存泄漏
避免全局变量：全局变量的引用不会被释放
使用上下文管理器：确保资源正确释放

理解Python的内存管理机制能帮助你写出更高效的代码，避免常见的内存问题。

内存管理与垃圾回收

引用计数

引用计数的限制

循环引用与垃圾回收

GC分代机制

弱引用

弱引用字典

弱引用回调

内存泄漏检测

tracemalloc内存分析

详细内存分析

内存优化技巧

使用__slots__

使用生成器减少内存

内存管理最佳实践

使用slots