弱引用
Python弱引用详解：weakref模块、WeakKeyDictionary。
概述
弱引用（Weak Reference）是一种不增加对象引用计数的引用方式。当对象只剩下弱引用时，垃圾回收器可以正常回收该对象。Python 通过 weakref 模块提供弱引用支持，常用于缓存、观察者模式和避免循环引用等场景。
基础概念
强引用与弱引用
import weakref
class Data:
def __init__(self, value):
self.value = value
# 强引用：引用计数 +1，对象不会被回收
obj = Data(42)
# obj 持有强引用，对象不会被 GC 回收
# 弱引用：不增加引用计数，对象可以被正常回收
ref = weakref.ref(obj)
# 通过 ref() 访问对象
print(ref().value) # 42
# 删除强引用后，弱引用返回 None
del obj
print(ref()) # None — 对象已被回收弱引用的适用类型
并非所有对象都支持弱引用：
- 支持的：自定义类实例、函数、方法（某些情况）、类型对象
- 不支持的：list、dict、int、str、tuple 等内置类型
# 不支持弱引用的类型
# weakref.ref([1, 2, 3]) # TypeError
# weakref.ref(42) # TypeError
# weakref.ref("hello") # TypeError
# 可以通过子类化添加弱引用支持
class WeakList(list):
pass
wl = WeakList([1, 2, 3])
ref = weakref.ref(wl) # OK快速上手
基本弱引用
import weakref
class Resource:
def __init__(self, name):
self.name = name
def __repr__(self):
return f"Resource({self.name})"
# 创建弱引用
resource = Resource("数据库连接")
ref = weakref.ref(resource)
# 通过弱引用访问对象
print(ref()) # Resource(数据库连接)
# 检查对象是否还存活
if ref() is not None:
print(f"对象仍存活: {ref().name}")弱引用回调
当弱引用指向的对象被回收时，可以触发回调函数：
def on_destroy(ref):
"""对象被回收时的回调"""
print("对象已被回收")
obj = Resource("临时资源")
ref = weakref.ref(obj, on_destroy)
del obj # 输出: 对象已被回收详细用法
WeakKeyDictionary
键为弱引用的字典。当键对象被回收时，对应的条目自动删除：
class Node:
def __init__(self, name):
self.name = name
# 使用弱键字典跟踪节点的元数据
metadata = weakref.WeakKeyDictionary()
node1 = Node("A")
node2 = Node("B")
metadata[node1] = {"visited": False, "weight": 1.0}
metadata[node2] = {"visited": True, "weight": 2.0}
print(metadata[node1]) # {'visited': False, 'weight': 1.0}
# 删除 node1 后，对应条目自动删除
del node1
print(dict(metadata)) # 只剩 node2 的条目WeakValueDictionary
值为弱引用的字典。当值对象被回收时，对应的条目自动删除：
# 缓存场景：值是弱引用
cache = weakref.WeakValueDictionary()
class ExpensiveObject:
def __init__(self, key):
self.key = key
obj = ExpensiveObject("result")
cache["query_1"] = obj
print(cache["query_1"].key) # result
# 删除强引用后，缓存条目自动清除
del obj
print("query_1" in cache) # FalseWeakSet
元素为弱引用的集合。当元素被回收时自动从集合中移除：
# 跟踪所有活跃的连接
active_connections = weakref.WeakSet()
class Connection:
def __init__(self, id):
self.id = id
conn1 = Connection(1)
conn2 = Connection(2)
active_connections.add(conn1)
active_connections.add(conn2)
print(len(active_connections)) # 2
del conn1
print(len(active_connections)) # 1 — 自动移除弱引用方法
class Observer:
def __init__(self, name):
self.name = name
class Subject:
def __init__(self):
self._observers = weakref.WeakSet()
def register(self, observer):
self._observers.add(observer)
def notify(self, message):
for observer in self._observers:
print(f"通知 {observer.name}: {message}")
subject = Subject()
obs1 = Observer("观察者1")
obs2 = Observer("观察者2")
subject.register(obs1)
subject.register(obs2)
subject.notify("事件发生")
# 删除 obs1 后，不再收到通知
del obs1
subject.notify("另一个事件") # 只通知观察者2常见场景
场景一：对象缓存
class ObjectCache:
"""基于弱引用的对象缓存"""
def __init__(self):
self._cache = weakref.WeakValueDictionary()
def get_or_create(self, key, factory):
"""获取缓存对象，不存在则创建"""
obj = self._cache.get(key)
if obj is None:
obj = factory()
self._cache[key] = obj
return obj
cache = ObjectCache()
# 第一次创建
result = cache.get_or_create("query_1", lambda: ExpensiveObject("query_1"))
# 第二次从缓存获取
result2 = cache.get_or_create("query_1", lambda: ExpensiveObject("query_1"))
print(result is result2) # True — 同一个对象
# 删除所有强引用后缓存自动清除
del result, result2场景二：避免循环引用
class Parent:
def __init__(self):
self.children = []
class Child:
def __init__(self, parent):
# 使用弱引用避免循环引用
self._parent_ref = weakref.ref(parent)
parent.children.append(self)
@property
def parent(self):
p = self._parent_ref()
if p is None:
raise RuntimeError("父对象已被回收")
return p
# 创建父子关系
parent = Parent()
child = Child(parent)
# 删除父对象后，弱引用返回 None
del parent
# child.parent # RuntimeError: 父对象已被回收场景三：单例注册表
class SingletonRegistry:
"""弱引用单例注册表"""
_instances = weakref.WeakValueDictionary()
@classmethod
def get(cls, key, factory):
"""获取或创建单例"""
instance = cls._instances.get(key)
if instance is None:
instance = factory()
cls._instances[key] = instance
return instance
@classmethod
def clear(cls):
"""清除所有单例（测试用）"""
cls._instances.clear()注意事项
- 弱引用不适用于内置不可变类型（int、str、tuple 等）
- 通过弱引用访问对象时，必须检查返回值是否为 None
- WeakKeyDictionary 的键必须是可哈希且支持弱引用的对象
- 弱引用回调中不要创建对被回收对象的新的强引用
- 在多线程环境中使用弱引用容器时需要注意同步
- 弱引用不是解决所有内存泄漏的银弹，应先分析实际的引用关系
进阶用法
finalize 自动清理
# weakref.finalize 在对象被回收时自动调用清理函数
class TempFile:
def __init__(self, path):
self.path = path
self._finalizer = weakref.finalize(self, self._cleanup, path)
@staticmethod
def _cleanup(path):
"""对象被回收时自动删除临时文件"""
import os
if os.path.exists(path):
os.remove(path)
print(f"已清理临时文件: {path}")
def close(self):
"""手动清理"""
self._finalizer()
# 使用
tmp = TempFile("/tmp/data.txt")
del tmp # 自动调用 _cleanup弱引用与描述符配合
class ObservableAttribute:
"""可观察的属性描述符"""
def __init__(self, name):
self.name = name
self._observers = weakref.WeakKeyDictionary()
def __set_name__(self, owner, name):
self.name = name
self.storage_name = f"_{name}"
def __get__(self, obj, objtype=None):
if obj is None:
return self
return getattr(obj, self.storage_name, None)
def __set__(self, obj, value):
old = getattr(obj, self.storage_name, None)
setattr(obj, self.storage_name, value)
if old != value:
self._notify(obj, old, value)
def observe(self, obj, callback):
"""注册观察者"""
if obj not in self._observers:
self._observers[obj] = []
self._observers[obj].append(callback)
def _notify(self, obj, old, new):
for callback in self._observers.get(obj, []):
callback(old, new)弱引用与 slots
# 使用 __slots__ 的类默认支持弱引用
# 但如果 __slots__ 中没有 __weakref__，则不支持
class WithWeakRef:
__slots__ = ('value', '__weakref__')
def __init__(self, value):
self.value = value
obj = WithWeakRef(42)
ref = weakref.ref(obj) # OK
class WithoutWeakRef:
__slots__ = ('value',) # 没有 __weakref__
# ref = weakref.ref(WithoutWeakRef(42)) # TypeError