Python进阶-内存管理详解

2019-02-15 | 阅读：次

Python内存管理的重要性

Python的内存管理机制是Python语言的核心特性之一。理解Python的内存管理原理对于编写高效、内存友好的Python代码至关重要。Python使用引用计数和垃圾回收相结合的方式来管理内存。

引用计数机制

1. 基础引用计数

import sys

def demonstrate_ref_count():
    """演示引用计数机制"""
    print("=== 引用计数基础演示 ===")
    
    # 创建对象
    a = [1, 2, 3]
    print(f"创建列表a，引用计数: {sys.getrefcount(a)}")
    
    # 增加引用
    b = a
    print(f"b = a 后，引用计数: {sys.getrefcount(a)}")
    
    # 添加到容器中
    container = [a]
    print(f"添加到容器后，引用计数: {sys.getrefcount(a)}")
    
    # 函数参数也会增加引用计数
    def func(param):
        print(f"函数内部，引用计数: {sys.getrefcount(param)}")
        return param
    
    result = func(a)
    print(f"函数调用后，引用计数: {sys.getrefcount(a)}")
    
    # 删除引用
    del b
    print(f"删除b后，引用计数: {sys.getrefcount(a)}")
    
    del container[0]
    print(f"从容器删除后，引用计数: {sys.getrefcount(a)}")
    
    # 最后删除
    del a
    print("删除a后，对象被回收")

demonstrate_ref_count()

2. 循环引用问题

import sys
import gc

class Node:
    """节点类 - 演示循环引用"""
    def __init__(self, value):
        self.value = value
        self.next = None
        self.prev = None
    
    def __repr__(self):
        return f"Node({self.value})"
    
    def __del__(self):
        print(f"节点 {self.value} 被删除")

def create_circular_reference():
    """创建循环引用"""
    print("\n=== 循环引用演示 ===")
    
    # 创建节点
    node1 = Node(1)
    node2 = Node(2)
    node3 = Node(3)
    
    print(f"创建节点后，引用计数:")
    print(f"node1: {sys.getrefcount(node1)}")
    print(f"node2: {sys.getrefcount(node2)}")
    print(f"node3: {sys.getrefcount(node3)}")
    
    # 创建循环引用
    node1.next = node2
    node2.prev = node1
    node2.next = node3
    node3.prev = node2
    node3.next = node1  # 形成循环
    node1.prev = node3
    
    print(f"\n创建循环引用后，引用计数:")
    print(f"node1: {sys.getrefcount(node1)}")
    print(f"node2: {sys.getrefcount(node2)}")
    print(f"node3: {sys.getrefcount(node3)}")
    
    # 删除局部引用
    del node1, node2, node3
    print("\n删除局部引用后:")
    print("对象仍然存在（循环引用导致）")
    
    # 手动触发垃圾回收
    print("\n手动触发垃圾回收:")
    collected = gc.collect()
    print(f"回收了 {collected} 个对象")

create_circular_reference()

垃圾回收机制

1. 分代垃圾回收

import gc
import sys

def demonstrate_garbage_collection():
    """演示垃圾回收机制"""
    print("\n=== 垃圾回收机制演示 ===")
    
    # 查看当前垃圾回收统计
    print("当前垃圾回收统计:")
    stats = gc.get_stats()
    for i, stat in enumerate(stats):
        print(f"代 {i}: {stat}")
    
    # 查看垃圾回收阈值
    print(f"\n垃圾回收阈值: {gc.get_threshold()}")
    
    # 创建大量临时对象
    print("\n创建大量临时对象...")
    temp_objects = []
    for i in range(1000):
        temp_objects.append([i, i*2, i*3])
    
    print(f"创建了 {len(temp_objects)} 个临时对象")
    
    # 删除引用
    del temp_objects
    print("删除引用后")
    
    # 手动触发垃圾回收
    print("\n手动触发垃圾回收:")
    collected = gc.collect()
    print(f"回收了 {collected} 个对象")
    
    # 查看回收后的统计
    print("\n回收后统计:")
    stats_after = gc.get_stats()
    for i, stat in enumerate(stats_after):
        print(f"代 {i}: {stat}")

demonstrate_garbage_collection()

2. 垃圾回收配置

import gc

def configure_garbage_collection():
    """配置垃圾回收"""
    print("\n=== 垃圾回收配置 ===")
    
    # 获取当前配置
    print(f"当前阈值: {gc.get_threshold()}")
    print(f"垃圾回收是否启用: {gc.isenabled()}")
    
    # 设置新的阈值
    print("\n设置新的阈值...")
    gc.set_threshold(700, 10, 10)  # (第0代, 第1代, 第2代)
    print(f"新阈值: {gc.get_threshold()}")
    
    # 禁用垃圾回收
    print("\n禁用垃圾回收:")
    gc.disable()
    print(f"垃圾回收是否启用: {gc.isenabled()}")
    
    # 重新启用
    print("\n重新启用垃圾回收:")
    gc.enable()
    print(f"垃圾回收是否启用: {gc.isenabled()}")
    
    # 恢复默认阈值
    gc.set_threshold(700, 10, 10)
    print(f"恢复默认阈值: {gc.get_threshold()}")

configure_garbage_collection()

弱引用

1. 基础弱引用

import weakref
import sys

class Node:
    """节点类 - 使用弱引用避免循环引用"""
    def __init__(self, value):
        self.value = value
        self.children = weakref.WeakSet()  # 使用弱引用集合
        self.parent = None

    def add_child(self, child):
        """添加子节点"""
        child.parent = weakref.ref(self)  # 使用弱引用
        self.children.add(child)
        print(f"节点 {self.value} 添加子节点 {child.value}")
    
    def get_parent(self):
        """获取父节点"""
        if self.parent is not None:
            parent = self.parent()
            if parent is not None:
                return parent
            else:
                print(f"节点 {self.value} 的父节点已被回收")
                self.parent = None
        return None

    def __repr__(self):
        return f"Node({self.value})"
    
    def __del__(self):
        print(f"节点 {self.value} 被删除")

def demonstrate_weak_references():
    """演示弱引用"""
    print("\n=== 弱引用演示 ===")
    
    # 创建节点
    root = Node("root")
    child1 = Node("child1")
    child2 = Node("child2")
    
    print(f"创建节点后，引用计数:")
    print(f"root: {sys.getrefcount(root)}")
    print(f"child1: {sys.getrefcount(child1)}")
    print(f"child2: {sys.getrefcount(child2)}")
    
    # 建立父子关系（使用弱引用）
    root.add_child(child1)
    root.add_child(child2)
    
    print(f"\n建立关系后，引用计数:")
    print(f"root: {sys.getrefcount(root)}")
    print(f"child1: {sys.getrefcount(child1)}")
    print(f"child2: {sys.getrefcount(child2)}")
    
    # 测试访问父节点
    parent = child1.get_parent()
    print(f"child1的父节点: {parent}")
    
    # 删除根节点引用
    print("\n删除根节点引用:")
    del root

    # 强制垃圾回收
    import gc
    gc.collect()
    
    # 尝试访问父节点（应该返回None）
    parent = child1.get_parent()
    print(f"child1的父节点: {parent}")
    
    # 清理
    del child1, child2
    gc.collect()

demonstrate_weak_references()

2. 弱引用回调

import weakref

class DataManager:
    """数据管理器 - 演示弱引用回调"""
    def __init__(self):
        self.data_objects = {}
        self.callback_count = 0

    def register_data(self, key, data_obj):
        """注册数据对象"""
        def cleanup_callback(ref):
            """清理回调函数"""
            self.callback_count += 1
            print(f"数据对象 {key} 被回收，回调计数: {self.callback_count}")
            # 清理相关资源
            if key in self.data_objects:
                del self.data_objects[key]
        
        # 创建弱引用并设置回调
        weak_ref = weakref.ref(data_obj, cleanup_callback)
        self.data_objects[key] = weak_ref
        print(f"注册数据对象: {key}")
    
    def get_data(self, key):
        """获取数据对象"""
        if key in self.data_objects:
            weak_ref = self.data_objects[key]
            data_obj = weak_ref()
            if data_obj is not None:
                return data_obj
            else:
                print(f"数据对象 {key} 已被回收")
                del self.data_objects[key]
        return None

    def list_objects(self):
        """列出所有对象"""
        print(f"当前管理的对象: {list(self.data_objects.keys())}")
        for key, weak_ref in self.data_objects.items():
            obj = weak_ref()
            status = "存在" if obj is not None else "已回收"
            print(f"  {key}: {status}")

def demonstrate_weakref_callback():
    """演示弱引用回调"""
    print("\n=== 弱引用回调演示 ===")
    
    manager = DataManager()
    
    # 创建数据对象
    data1 = [1, 2, 3]
    data2 = {"name": "test", "value": 42}
    data3 = "hello world"
    
    # 注册数据
    manager.register_data("list_data", data1)
    manager.register_data("dict_data", data2)
    manager.register_data("str_data", data3)
    
    # 查看对象状态
    manager.list_objects()
    
    # 访问数据
    print(f"\n访问数据:")
    print(f"list_data: {manager.get_data('list_data')}")
    print(f"dict_data: {manager.get_data('dict_data')}")
    
    # 删除数据引用
    print(f"\n删除数据引用:")
    del data1, data3

    # 强制垃圾回收
    import gc
    gc.collect()
    
    # 查看对象状态
    manager.list_objects()
    
    # 清理
    del data2
    gc.collect()
    manager.list_objects()

demonstrate_weakref_callback()

内存优化策略

1. 内存池和对象复用

import sys
from array import array

def memory_optimization_demo():
    """内存优化演示"""
    print("\n=== 内存优化策略 ===")
    
    # 1. 使用__slots__减少内存占用
    class RegularClass:
        def __init__(self, x, y):
            self.x = x
            self.y = y

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

    # 比较内存占用
    regular = RegularClass(1, 2)
    slots = SlotsClass(1, 2)
    
    print(f"普通类实例大小: {sys.getsizeof(regular)} 字节")
    print(f"__slots__类实例大小: {sys.getsizeof(slots)} 字节")
    
    # 2. 使用生成器节省内存
    def regular_list(n):
        """创建普通列表"""
        return [i*i for i in range(n)]
    
    def generator_list(n):
        """创建生成器"""
        return (i*i for i in range(n))
    
    # 比较内存占用
    n = 1000
    regular_data = regular_list(n)
    generator_data = generator_list(n)
    
    print(f"\n普通列表大小: {sys.getsizeof(regular_data)} 字节")
    print(f"生成器大小: {sys.getsizeof(generator_data)} 字节")
    
    # 3. 使用array模块
    python_list = [1, 2, 3, 4, 5]
    int_array = array('i', [1, 2, 3, 4, 5])
    
    print(f"\nPython列表大小: {sys.getsizeof(python_list)} 字节")
    print(f"array数组大小: {sys.getsizeof(int_array)} 字节")

memory_optimization_demo()

2. 内存泄漏检测

import gc
import tracemalloc
import sys

def detect_memory_leaks():
    """内存泄漏检测"""
    print("\n=== 内存泄漏检测 ===")
    
    # 启动内存跟踪
    tracemalloc.start()
    
    # 记录初始内存状态
    snapshot1 = tracemalloc.take_snapshot()
    
    # 模拟可能的内存泄漏
    leaked_objects = []
    
    def create_leaked_data():
        """创建可能泄漏的数据"""
        data = []
        for i in range(1000):
            data.append({
                'id': i,
                'data': [j for j in range(100)],
                'timestamp': '2019-02-15T10:30:00Z'
            })
        return data

    # 创建数据但不清理
    leaked_objects.append(create_leaked_data())
    leaked_objects.append(create_leaked_data())
    
    # 记录内存状态
    snapshot2 = tracemalloc.take_snapshot()
    
    # 分析内存差异
    top_stats = snapshot2.compare_to(snapshot1, 'lineno')
    
    print("内存使用统计 (前10个):")
    for index, stat in enumerate(top_stats[:10], 1):
        print(f"{index}. {stat}")
    
    # 获取当前内存使用
    current, peak = tracemalloc.get_traced_memory()
    print(f"\n当前内存使用: {current / 1024 / 1024:.2f} MB")
    print(f"峰值内存使用: {peak / 1024 / 1024:.2f} MB")
    
    # 清理泄漏的对象
    print("\n清理泄漏对象...")
    del leaked_objects
    gc.collect()
    
    # 记录清理后的内存状态
    snapshot3 = tracemalloc.take_snapshot()
    current_after, peak_after = tracemalloc.get_traced_memory()
    
    print(f"清理后内存使用: {current_after / 1024 / 1024:.2f} MB")
    print(f"内存释放: {(current - current_after) / 1024 / 1024:.2f} MB")
    
    # 停止内存跟踪
    tracemalloc.stop()

detect_memory_leaks()

实际应用案例

1. 缓存系统

import weakref
import time
from typing import Any, Dict, Optional

class MemoryEfficientCache:
    """内存高效的缓存系统"""
    def __init__(self, max_size: int = 1000):
        self.max_size = max_size
        self.cache: Dict[str, Any] = {}
        self.access_times: Dict[str, float] = {}
        self.weak_refs: Dict[str, weakref.ref] = {}
    
    def get(self, key: str) -> Optional[Any]:
        """获取缓存项"""
        if key in self.cache:
            # 更新访问时间
            self.access_times[key] = time.time()
            return self.cache[key]
        return None

    def put(self, key: str, value: Any) -> None:
        """添加缓存项"""
        # 检查缓存大小
        if len(self.cache) >= self.max_size:
            self._evict_oldest()
        
        self.cache[key] = value
        self.access_times[key] = time.time()
                                
        # 为大型对象创建弱引用
        if sys.getsizeof(value) > 1024:  # 大于1KB的对象
            self.weak_refs[key] = weakref.ref(value)
    
    def _evict_oldest(self) -> None:
        """移除最旧的缓存项"""
        if not self.access_times:
            return

        # 找到最旧的项
        oldest_key = min(self.access_times.keys(), 
                        key=lambda k: self.access_times[k])
        
        print(f"移除最旧缓存项: {oldest_key}")
        self.remove(oldest_key)
    
    def remove(self, key: str) -> None:
        """移除缓存项"""
        self.cache.pop(key, None)
        self.access_times.pop(key, None)
        self.weak_refs.pop(key, None)
    
    def cleanup_dead_refs(self) -> None:
        """清理已回收的弱引用"""
        dead_keys = []
        for key, weak_ref in self.weak_refs.items():
            if weak_ref() is None:
                dead_keys.append(key)
        
        for key in dead_keys:
            print(f"清理已回收的弱引用: {key}")
            self.cache.pop(key, None)
            self.access_times.pop(key, None)
            del self.weak_refs[key]
    
    def get_stats(self) -> Dict[str, Any]:
        """获取缓存统计"""
        total_size = sum(sys.getsizeof(v) for v in self.cache.values())
        return {
            'size': len(self.cache),
            'max_size': self.max_size,
            'total_memory': total_size,
            'weak_refs': len(self.weak_refs)
        }

def test_memory_efficient_cache():
    """测试内存高效缓存"""
    print("\n=== 内存高效缓存测试 ===")
    
    cache = MemoryEfficientCache(max_size=3)
    
    # 添加缓存项
    cache.put("small", "hello")
    cache.put("medium", [i for i in range(100)])
    cache.put("large", [i for i in range(1000)])
    
    print("缓存统计:", cache.get_stats())
    
    # 测试获取
    print(f"获取'small': {cache.get('small')}")
    print(f"获取'medium': {cache.get('medium')}")
    
    # 添加新项（会触发清理）
    cache.put("new_item", "new_value")
    print("添加新项后的统计:", cache.get_stats())
    
    # 清理测试
    cache.cleanup_dead_refs()
    print("清理后的统计:", cache.get_stats())

test_memory_efficient_cache()

总结

Python内存管理的关键要点：

引用计数：Python的主要内存管理机制
垃圾回收：处理循环引用和分代回收
弱引用：避免循环引用，支持回调
内存优化：使用__slots__、生成器、array等
内存监控：使用tracemalloc检测内存使用
实际应用：缓存系统、资源管理等场景

掌握Python内存管理机制，可以编写更高效、更稳定的Python应用程序。

转载请注明：周志洋的博客 » Python进阶-内存管理详解