Python 弱引用与垃圾回收：缓存设计与内存泄漏排查

1
class Node:
2
    def __init__(self, name):
3
        self.name = name
4
        self.parent = None
5
        self.children = []
6

7
    def add_child(self, child):
8
        child.parent = self
9
        self.children.append(child)
10

11
root = Node("root")
12
leaf = Node("leaf")
13
root.add_child(leaf)
14

15
del root, leaf  # 看起来都删了，但 root 和 leaf 互相引用
16
print(gc.collect())  # 手动触发 GC，输出: 2 (回收了 2 个对象)

1
import weakref
2

3
class User:
4
    def __init__(self, name):
5
        self.name = name
6
        print(f"User {name} created")
7

8
cache = weakref.WeakValueDictionary()
9

10
def get_user(name):
11
    if name not in cache:
12
        cache[name] = User(name)
13
    return cache[name]
14

15
u = get_user("alice")  # User alice created
16
print(len(cache))       # 1
17
del u
18
import gc; gc.collect() # 确保触发 GC
19
print(len(cache))       # 0  alice 自动从缓存中消失！

1
import weakref
2
import time
3

4
class Connection:
5
    def __init__(self, db_name):
6
        self.db_name = db_name
7
        self.created_at = time.time()
8

9
    def query(self, sql):
10
        return f"Query '{sql}' on {self.db_name}"
11

12
class ConnectionPool:
13
    def __init__(self):
14
        self._pool = weakref.WeakValueDictionary()
15

16
    def get_connection(self, db_name):
17
        if db_name not in self._pool:
18
            # 实际场景中这里会创建真正的连接
19
            self._pool[db_name] = Connection(db_name)
20
        return self._pool[db_name]
21

22
pool = ConnectionPool()
23
conn = pool.get_connection("orders_db")
24
conn.query("SELECT * FROM orders")  # Query 'SELECT * FROM orders' on orders_db
25
del conn
26
import gc; gc.collect()
27
print(len(pool._pool))  # 0  连接被自动清理

1
import weakref
2

3
class GraphNode:
4
    def __init__(self, name):
5
        self.name = name
6

7
# 给节点附加一些元数据
8
metadata = weakref.WeakKeyDictionary()
9

10
node = GraphNode("A")
11
metadata[node] = {"color": "red", "weight": 1.0}
12
print(metadata[node])  # {'color': 'red', 'weight': 1.0}
13

14
del node
15
import gc; gc.collect()
16
print(len(metadata))   # 0  节点被回收，metadata 自动清理

1
import weakref
2

3
class EventEmitter:
4
    def __init__(self):
5
        self._listeners = weakref.WeakSet()
6

7
    def add_listener(self, listener):
8
        self._listeners.add(listener)
9

10
    def emit(self, event):
11
        for listener in list(self._listeners):
12
            listener.on_event(event)
13

14
class Listener:
15
    def __init__(self, name):
16
        self.name = name
17

18
    def on_event(self, event):
19
        print(f"{self.name} received: {event}")
20

21
emitter = EventEmitter()
22
l1 = Listener("L1")
23
l2 = Listener("L2")
24
emitter.add_listener(l1)
25
emitter.add_listener(l2)
26

27
emitter.emit("hello")  # L1 received: hello
28
                          # L2 received: hello
29

30
del l1
31
emitter.emit("bye")    # L2 received: bye  (L1 自动被移除)

1
import weakref
2
import os
3

4
def cleanup_file(path, fd):
5
    print(f"清理文件: {path}")
6
    fd.close()
7

8
class TempFile:
9
    def __init__(self, path):
10
        self.path = path
11
        self.fd = open(path, 'w')
12
        # 注册终结器
13
        self._finalizer = weakref.finalize(
14
            self, cleanup_file, self.path, self.fd
15
        )
16

17
t = TempFile("/tmp/test.txt")
18
t.fd.write("hello")
19
del t
20
import gc; gc.collect()  # 输出: 清理文件: /tmp/test.txt

1
import weakref
2
import threading
3

4
class ResourceManager:
5
    """模拟一个资源管理器，确保资源最终被释放"""
6
    def __init__(self, name):
7
        self.name = name
8
        self.lock = threading.Lock()
9
        weakref.finalize(
10
            self, self._cleanup,
11
            self.name, self.lock
12
        )
13

14
    @staticmethod
15
    def _cleanup(name, lock):
16
        # 注意：必须用 staticmethod，否则 self 会被持有
17
        # 导致对象永远不能被 GC
18
        print(f"释放资源: {name}")
19

20
r = ResourceManager("db-conn-1")
21
del r
22
import gc; gc.collect()  # 输出: 释放资源: db-conn-1

1
import os
2
import psutil
3

4
def get_memory_mb():
5
    process = psutil.Process(os.getpid())
6
    return process.memory_info().rss / 1024 / 1024
7

8
print(f"初始内存: {get_memory_mb():.1f} MB")
9

10
# 模拟泄漏
11
leak = []
12
for i in range(1000):
13
    leak.append({f"key_{j}": "x" * 1000 for j in range(100)})
14

15
print(f"泄漏后: {get_memory_mb():.1f} MB")  # 显著增加

1
import gc
2
from collections import Counter
3

4
# 创建大量临时对象
5
def process_data():
6
    return [Data() for _ in range(1000)]
7

8
class Data:
9
    pass
10

11
# 拍摄快照
12
gc.collect()
13
before = gc.get_objects()
14
print(f"快照前对象数: {len(before)}")
15

16
process_data()
17
import gc; gc.collect()
18

19
after = gc.get_objects()
20
print(f"快照后对象数: {len(after)}")  # 应该差不多
21

22
# 找出新增的对象类型
23
new_objects = [id(o) for o in after]
24
old_ids = set(id(o) for o in before)
25
diff = [o for o in after if id(o) not in old_ids]
26
counter = Counter(type(o).__name__ for o in diff)
27
print("新增对象类型分布:", counter.most_common(5))

1
import gc
2

3
# 开启循环引用调试（注意：会影响性能）
4
gc.set_debug(gc.DEBUG_COLLECTABLE | gc.DEBUG_UNCOLLECTABLE | gc.DEBUG_INSTANCES | gc.DEBUG_OBJECTS)
5

6
class Node:
7
    def __init__(self, name):
8
        self.name = name
9
        self.next = None
10
    def __repr__(self):
11
        return f"Node({self.name})"
12

13
# 创建循环引用
14
a = Node("A")
15
b = Node("B")
16
a.next = b
17
b.next = a  # 循环引用！
18

19
del a, b
20
collected = gc.collect()  # 会打印出无法回收的对象
21
print(f"回收: {collected} 个对象")
22

23
# 关闭调试
24
gc.set_debug(0)

1
# pip install objgraph
2
import objgraph
3

4
class A:
5
    def __init__(self):
6
        self.b = None
7

8
class B:
9
    def __init__(self):
10
        self.a = None
11

12
a = A()
13
b = B()
14
a.b = b
15
b.a = a
16

17
# 展示最常见的类型
18
objgraph.show_most_common_types(limit=10)
19

20
# 展示 a 对象的引用链
21
objgraph.show_backrefs(a, filename="a_refs.png")
22

23
# 展示引用关系图
24
objgraph.show_refs([a, b], filename="refs.png")

1
import tracemalloc
2

3
tracemalloc.start()
4

5
# 模拟泄漏
6
big_strings = []
7
for i in range(100):
8
    big_strings.append("x" * 100000)
9

10
# 获取当前内存快照
11
snapshot = tracemalloc.take_snapshot()
12
top_stats = snapshot.statistics("lineno")
13

14
print("Top 10 内存分配位置:")
15
for stat in top_stats[:10]:
16
    print(stat)

1
import weakref
2

3
class Foo: pass
4

5
foo = Foo()
6
ref = weakref.ref(foo)
7

8
callbacks = [lambda: ref()]
9
print(callbacks[0]())  # ✅ <Foo object>
10

11
# 但如果这样：
12
import gc; gc.collect()
13
del foo
14
print(callbacks[0]())  # ✅ None (lambda 没有持有 foo 的强引用)
15

16
# ❌ 错误：直接捕获对象
17
callbacks_bad = [lambda: foo]
18
del foo
19
import gc; gc.collect()
20
print(callbacks_bad[0]())  # ❌ NameError (foo 已删除)

1
import weakref
2

3
class Bad:
4
    def __init__(self):
5
        # ❌ 这样写，self 会被 self._finalizer 持有
6
        # 导致对象永远不能被 GC
7
        self._finalizer = weakref.finalize(
8
            self, self.cleanup  # self 间接持有 self
9
        )
10

11
    def cleanup(self):
12
        print("cleanup")
13

14
# ✅ 正确写法
15
class Good:
16
    def __init__(self):
17
        self._finalizer = weakref.finalize(
18
            self, self._cleanup, id(self)
19
        )
20

21
    @staticmethod
22
    def _cleanup(obj_id):
23
        print(f"cleanup obj {obj_id}")

1
import weakref
2

3
# ❌ 这些都不支持
4
weakref.ref(1)         # TypeError
5
weakref.ref("a")       # TypeError
6
weakref.ref([1, 2])    # TypeError
7
weakref.ref((1, 2))    # TypeError
8

9
# ✅ 解决：包一层 class
10
class Box:
11
    __slots__ = ("value",)
12
    def __init__(self, v):
13
        self.value = v
14

15
ref = weakref.ref(Box("hello"))

1
import weakref
2
import threading
3

4
cache = weakref.WeakValueDictionary()
5

6
# ❌ 危险：不是线程安全的
7
def worker():
8
    for i in range(1000):
9
        cache[i] = SomeObject(i)
10
        _ = cache.get(i)
11

12
# ✅ 解决：加锁
13
class ThreadSafeWeakCache:
14
    def __init__(self):
15
        self._cache = weakref.WeakValueDictionary()
16
        self._lock = threading.RLock()
17

18
    def get_or_create(self, key, factory):
19
        with self._lock:
20
            if key in self._cache:
21
                return self._cache[key]
22
            obj = factory()
23
            self._cache[key] = obj
24
            return obj