Python 内存管理

# Python 3.10+
import sys

# 创建一个列表对象
data = [1, 2, 3]
print(sys.getrefcount(data))  # 输出: 2
# 为什么是 2？因为 getrefcount 的参数本身也是一个临时引用

# 增加引用
ref = data
print(sys.getrefcount(data))  # 输出: 3

# 放入容器
container = [data]
print(sys.getrefcount(data))  # 输出: 4

# 减少引用
del ref
print(sys.getrefcount(data))  # 输出: 3

container.clear()
print(sys.getrefcount(data))  # 输出: 2

# Python 3.10+
class Node:
    def __init__(self, name: str) -> None:
        self.name = name
        self.next: Node | None = None
        self.prev: Node | None = None

# 创建循环引用
a = Node("A")
b = Node("B")
a.next = b
b.prev = a

# 删除外部引用
del a
del b
# 此时 a 和 b 互相引用，引用计数各为 1，不会归零
# 但分代垃圾回收器会在后续扫描中检测并回收它们

# Python 3.10+
import gc
import sys

# 查看各代阈值
print(gc.get_threshold())  # 典型输出: (700, 10, 10)
# 含义: 第 0 代对象数超过 700 时触发 0 代回收
#       每 10 次 0 代回收触发 1 次 1 代回收
#       每 10 次 1 代回收触发 1 次 2 代回收

# 查看各代对象数量
print(gc.get_count())  # 例如: (345, 5, 1)

# 手动触发回收
collected = gc.collect()  # 默认执行第 2 代（全量）回收
print(f"回收了 {collected} 个对象")

# 查看不可达但无法释放的对象（通常是因为定义了 __del__）
print(gc.garbage)

# 调整阈值（谨慎使用）
gc.set_threshold(1000, 15, 15)

# 禁用/启用自动垃圾回收
gc.disable()
# ... 执行内存敏感操作 ...
gc.enable()

# Python 3.10+
import sys


class LifecycleDemo:
    def __new__(cls, name: str) -> "LifecycleDemo":
        print(f"[__new__] 为 {name} 分配内存")
        instance = super().__new__(cls)
        return instance

    def __init__(self, name: str) -> None:
        print(f"[__init__] 初始化 {name}")
        self.name = name

    def __del__(self) -> None:
        print(f"[__del__] {self.name} 被销毁")


obj = LifecycleDemo("demo")
# 输出:
# [__new__] 为 demo 分配内存
# [__init__] 初始化 demo

del obj
# 输出: [__del__] demo 被销毁

# Python 3.10+
import gc


class BadPattern:
    def __init__(self, name: str) -> None:
        self.name = name

    def __del__(self) -> None:
        # 危险：如果 self.other 已被回收，这里可能出错
        print(f"清理 {self.name}, 关联: {self.other.name}")


a = BadPattern("A")
b = BadPattern("B")
a.other = b
b.other = a  # 循环引用

del a, b
gc.collect()
print(gc.garbage)  # 包含 a 和 b，它们无法被自动回收

# Python 3.10+
import sys


class RegularPoint:
    def __init__(self, x: float, y: float) -> None:
        self.x = x
        self.y = y


class SlottedPoint:
    __slots__ = ("x", "y")

    def __init__(self, x: float, y: float) -> None:
        self.x = x
        self.y = y


rp = RegularPoint(1.0, 2.0)
sp = SlottedPoint(1.0, 2.0)

print(sys.getsizeof(rp))          # 约 56 字节
print(sys.getsizeof(rp.__dict__))  # 额外 104 字节（字典本身）
print(sys.getsizeof(sp))          # 约 48 字节（无 __dict__）

# 大规模创建时的差异
N = 1_000_000
regular_mem = N * (sys.getsizeof(rp) + sys.getsizeof(rp.__dict__))
slotted_mem = N * sys.getsizeof(sp)
print(f"常规类 {N} 个实例约占用: {regular_mem / 1024 / 1024:.1f} MB")
print(f"__slots__ 类 {N} 个实例约占用: {slotted_mem / 1024 / 1024:.1f} MB")

# Python 3.10+
import weakref
import gc


class Cache:
    """使用弱引用的简单缓存"""
    def __init__(self) -> None:
        self._cache: weakref.WeakValueDictionary[int, object] = (
            weakref.WeakValueDictionary()
        )

    def set(self, key: int, value: object) -> None:
        self._cache[key] = value

    def get(self, key: int) -> object | None:
        return self._cache.get(key)


class ExpensiveObject:
    def __init__(self, data: str) -> None:
        self.data = data * 10000  # 模拟大对象


cache = Cache()
obj = ExpensiveObject("hello")
cache.set(1, obj)

print(cache.get(1))  # 可以获取到

del obj  # 删除唯一强引用
gc.collect()

print(cache.get(1))  # None，对象已被自动清理

# Python 3.10+
from collections import deque
from typing import TypeVar

T = TypeVar("T")


class ObjectPool:
    """通用对象池"""
    def __init__(self, factory, max_size: int = 100) -> None:
        self._pool: deque = deque()
        self._factory = factory
        self._max_size = max_size

    def acquire(self):
        if self._pool:
            return self._pool.pop()
        return self._factory()

    def release(self, obj) -> None:
        if len(self._pool) < self._max_size:
            self._pool.append(obj)
        # 超过最大容量时直接丢弃，让 GC 回收


# 使用示例：复用大缓冲区
import array
pool = ObjectPool(lambda: array.array("b", [0] * 4096), max_size=50)

buf = pool.acquire()
# ... 使用 buf ...
pool.release(buf)  # 归还以便复用

# Python 3.10+
import sys

print(sys.getsizeof(42))           # 28 字节（小整数）
print(sys.getsizeof("hello"))      # 54 字节
print(sys.getsizeof([]))           # 56 字节（空列表）
print(sys.getsizeof([1, 2, 3]))    # 80 字节（注意：只计算容器本身，不含元素）

# Python 3.10+
import tracemalloc


def leak_simulation() -> None:
    """模拟内存泄漏：全局列表不断增长"""
    global _leaked
    _leaked.append("x" * 1024 * 1024)  # 每次追加 1MB


_leaked: list[str] = []

tracemalloc.start()

# 模拟多次操作
for i in range(10):
    leak_simulation()

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

# 显示内存占用最高的 5 个文件
print("=== 内存占用 Top 5 ===")
for stat in snapshot.statistics("lineno")[:5]:
    print(stat)

# 比较两个快照
tracemalloc.clear_traces()
snapshot_before = tracemalloc.take_snapshot()
for i in range(5):
    leak_simulation()
snapshot_after = tracemalloc.take_snapshot()

print("\n=== 增量对比 ===")
for stat in snapshot_after.compare_to(snapshot_before, "lineno")[:3]:
    print(stat)

tracemalloc.stop()

# Python 3.10+
# pip install memory-profiler

from memory_profiler import profile


@profile
def process_data() -> list[int]:
    """处理数据，观察每行内存变化"""
    data = [i for i in range(100_000)]        # 创建列表
    squared = [x ** 2 for x in data]          # 平方计算
    filtered = [x for x in squared if x > 50]  # 过滤
    return filtered


if __name__ == "__main__":
    process_data()

# Python 3.10+
# pip install objgraph
import objgraph

# 显示增长最快的对象类型
objgraph.show_growth(limit=5)

# 查找特定类型的对象并显示反向引用链
leaked_objects = objgraph.by_type("MyClass")
if leaked_objects:
    objgraph.show_backrefs(leaked_objects[0], max_depth=5, filename="backrefs.png")

# Python 3.10+
# 反模式：无界缓存
_cache: dict[str, object] = {}

def get_data(key: str) -> object:
    if key not in _cache:
        _cache[key] = expensive_computation(key)
    return _cache[key]
# 问题：_cache 永远不会清理，内存持续增长

# Python 3.10+
from functools import lru_cache


@lru_cache(maxsize=256)  # 限制最大条目数
def get_data(key: str) -> object:
    return expensive_computation(key)

# Python 3.10+
# 反模式：忘记关闭文件/连接
def read_all_files(paths: list[str]) -> list[str]:
    results = []
    for path in paths:
        f = open(path)  # 忘记关闭
        results.append(f.read())
    return results

# Python 3.10+
from pathlib import Path

def read_all_files(paths: list[str]) -> list[str]:
    results: list[str] = []
    for path in paths:
        with open(path) as f:
            results.append(f.read())
    return results

# Python 3.10+
# 反模式：异常时引用未被清理
def process() -> None:
    data = load_large_dataset()  # 加载大数据
    try:
        result = transform(data)
        save(result)
    except Exception:
        # data 仍然被 traceback 引用，无法释放
        raise

# Python 3.10+
def process() -> None:
    data = None
    try:
        data = load_large_dataset()
        result = transform(data)
        save(result)
    except Exception:
        raise
    finally:
        del data  # 显式释放引用

# 文件名: detect_cycle.py
"""使用 gc 模块检测和打破循环引用"""
import gc
import sys


class TreeNode:
    def __init__(self, value: int) -> None:
        self.value = value
        self.parent: TreeNode | None = None
        self.children: list[TreeNode] = []


# 创建一棵树
root = TreeNode(1)
child = TreeNode(2)
root.children.append(child)
child.parent = root  # 形成了 parent<->child 的循环引用

# 删除外部引用
del root, child

# 手动触发 GC
collected = gc.collect()
print(f"回收了 {collected} 个对象")  # 期望输出: 回收了 2 个对象

# 验证：不可达对象已被清理
print(gc.garbage)  # 期望输出: []

# 文件名: find_leak.py
"""使用 tracemalloc 定位内存泄漏源头"""
import tracemalloc
import time


class LeakyCache:
    """模拟一个会泄漏的缓存"""
    def __init__(self) -> None:
        self._data: dict[str, str] = {}

    def add(self, key: str, value: str) -> None:
        self._data[key] = value
        # BUG: 没有淘汰机制，缓存无限增长


def simulate() -> None:
    cache = LeakyCache()
    for i in range(100_000):
        cache.add(f"key_{i}", "x" * 1024)  # 每次约 1KB


tracemalloc.start()
snapshot_before = tracemalloc.take_snapshot()

simulate()

snapshot_after = tracemalloc.take_snapshot()

print("=== 内存增量 Top 5 ===")
for stat in snapshot_after.compare_to(snapshot_before, "lineno")[:5]:
    print(stat)
    # 期望输出: find_leak.py 中 LeakyCache.add 的行号会排在最前面

tracemalloc.stop()