Python弱引用与内存泄漏防治

Python弱引用与内存泄漏防治

weakref模块提供了创建弱引用的能力。弱引用不增加对象的引用计数，当对象只剩下弱引用时，GC可以回收它。

weakref.ref是最基础的弱引用：

import weakref

class ExpensiveObject:
def __init__(self, name):
self.name = name
self.data = bytearray(1024 * 1024) # 1MB数据

def __repr__(self):
return f"ExpensiveObject({self.name})"

obj = ExpensiveObject("test")
ref = weakref.ref(obj)

print(ref()) # ExpensiveObject(test)
del obj # 删除强引用
print(ref()) # None

ref()返回引用的对象，如果对象已被回收则返回None。ref()的返回值必须检查是否为None，否则后续操作会出错。

weakref的回调函数在对象被回收时触发：

def cleanup(ref):
print(f"对象被回收了: {ref}")

obj = ExpensiveObject("monitored")
ref = weakref.ref(obj, cleanup)

del obj # 输出: 对象被回收了:

回调在对象引用计数归零（生命周期结束）时立即调用，在对象的__del__之前执行。回调的接收参数是弱引用对象本身，不是原始对象（原始对象已不存在）。

weakref.proxy创建代理对象，可以直接访问原对象的方法和属性，但不需要额外的()调用：

obj = ExpensiveObject("proxy")
proxy = weakref.proxy(obj)

print(proxy.name) # "proxy"，直接访问属性
del obj
try:
print(proxy.name) # ReferenceError: weakly-referenced object no longer exists
except ReferenceError as e:
print(e)

proxy的缺点是访问时无法感知对象是否存活，只有实际访问时才会抛出ReferenceError。ref()则通过返回值是否为None明确告知对象状态。

WeakValueDictionary的典型应用是缓存：

import weakref

class ImageCache:
def __init__(self):
self._cache = weakref.WeakValueDictionary()

def get_image(self, path):
img = self._cache.get(path)
if img is not None:
print(f"缓存命中: {path}")
return img

print(f"加载图像: {path}")
img = Image(path)
self._cache[path] = img
return img

class Image:
def __init__(self, path):
self.path = path
self.data = f"image_data:{path}"

def render(self):
return f"rendering {self.data}"

cache = ImageCache()
img1 = cache.get_image("/photo.jpg")
img2 = cache.get_image("/photo.jpg") # 缓存命中

del img1 # 删除强引用
import gc
gc.collect()

img3 = cache.get_image("/photo.jpg") # 可能需要重新加载

WeakValueDictionary在值对象不再被外部持有时自动移除条目。不需要手动清理缓存中的过期条目。

但WeakValueDictionary有一个陷阱：键的存活期也依赖值。如果值已经被回收，键仍然存在，但访问返回None：

cache = weakref.WeakValueDictionary()
obj = ExpensiveObject("temp")
cache['key'] = obj
print(cache.get('key')) # ExpensiveObject(temp)
del obj
gc.collect()
print(cache.get('key')) # None

WeakKeyDictionary是另一种方向的弱引用字典：

class EventHandler:
def __init__(self):
self._handlers = weakref.WeakKeyDictionary()

def register(self, obj, callback):
self._handlers[obj] = callback

def notify(self, *args, **kwargs):
for obj, callback in list(self._handlers.items()):
if obj is not None:
callback(obj, *args, **kwargs)
# 字典自动清理已回收的对象

handler = EventHandler()
class Listener:
def on_event(self, data):
print(f"处理: {data}")

listener = Listener()
handler.register(listener, Listener.on_event)

handler.notify("test") # 调用listener.on_event
del listener
gc.collect()
handler.notify("test") # 不输出任何内容

WeakKeyDictionary在对象被回收时自动移除条目。常用于需要给对象附加元数据但又不想延长对象生命周期的场景。

weakref.WeakSet是弱引用集合：

class Service:
def __init__(self):
self._instances = weakref.WeakSet()

def track(self, obj):
self._instances.add(obj)

def active_count(self):
return len(self._instances)

service = Service()
for _ in range(10):
obj = ExpensiveObject("temp")
service.track(obj)
# obj在循环结束时可能被回收

import gc
gc.collect()
print(f"活跃实例: {service.active_count()}") # 可能少于10

WeakSet自动管理元素的生存期。元素被回收后自动从集合中移除，不需要手动清理。

使用finalize替代__del__进行资源清理：

import weakref
import tempfile
import os

class TempFileResource:
def __init__(self, suffix='.tmp'):
self.file = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
self.path = self.file.name

def cleanup(path):
try:
if os.path.exists(path):
os.unlink(path)
print(f"清理临时文件: {path}")
except OSError:
pass

weakref.finalize(self, cleanup, self.path)

rsc = TempFileResource()
print(f"创建临时文件: {rsc.path}")
del rsc # finalize立即执行清理

finalize比__del__更可靠：它在对象被回收时保证执行，不会因为循环引用而受阻。并且finalize可以接收参数，不像__del__只能操作实例属性。

finalize的另一个优势是支持调用优先级。通过atexit注册的函数在进程退出时也能执行，确保资源释放：

import atexit
import weakref

class DatabaseConnection:
def __init__(self, conn_string):
self.conn_string = conn_string
self._connection = None

def close_connection(conn_str):
print(f"关闭数据库连接: {conn_str}")

self._finalizer = weakref.finalize(self, close_connection, conn_string)
atexit.register(self._finalizer)

db = DatabaseConnection("postgresql://localhost/db")
# 即使忘记显式关闭，进程退出时也会清理

atexit注册的finalize在Python进程正常退出时执行。注意：如果程序被SIGKILL杀死，atexit处理程序不会运行。

weakref.getweakrefcount和getweakrefs检查对象的弱引用状态：

obj = ExpensiveObject("weak_demo")
ref1 = weakref.ref(obj)
ref2 = weakref.ref(obj)

print(weakref.getweakrefcount(obj)) # 2
print(len(weakref.getweakrefs(obj))) # 2

这个接口在调试时有用，可以确认对象被创建了多少弱引用。

循环引用中__del__的问题可以通过weakref解决：

class Node:
def __init__(self):
self.parent = None
self.children = []

def add_child(self, child):
self.children.append(child)
child.parent = weakref.ref(self) # 使用弱引用

def __del__(self):
print(f"Node deleted: {id(self)}")

parent = Node()
child = Node()
parent.add_child(child)

del parent
del child
gc.collect() # 两个节点都被正确回收

如果Node.parent是强引用，parent和child形成循环引用。使用weakref.ref(self)打破循环，GC可以正常回收。

weakref的代理模式在实现观察者模式时很实用：

class Observable:
def __init__(self):
self._observers = weakref.WeakSet()

def attach(self, observer):
self._observers.add(observer)

def detach(self, observer):
self._observers.discard(observer)

def notify(self, data):
dead = []
for observer in self._observers:
if observer() is not None:
observer(data)
else:
dead.append(observer)
for obs in dead:
self._observers.discard(obs)

class Widget:
def update(self, data):
print(f"Widget updated: {data}")

observable = Observable()
widget = Widget()
observable.attach(widget.update)

observable.notify("event1") # 正常通知
del widget
gc.collect()
observable.notify("event2") # widget已被回收

WeakSet中的回调函数在widget被回收后自动移除，不会尝试调用已销毁的对象。

Python 3.10中weakref的优化：ref对象的哈希操作速度提升了约40%，因为内部实现从Python层面移到了C层面，不再需要Python函数调用的开销。

内置类型对弱引用的支持情况：

import weakref

class WithSlots:
__slots__ = ('x', '__weakref__')

obj = WithSlots()
ref = weakref.ref(obj) # OK，因为__slots__包含了__weakref__

class WithoutWeak:
__slots__ = ('x',)

obj2 = WithoutWeak()
ref2 = weakref.ref(obj2) # TypeError: cannot create weak reference to 'WithoutWeak' object

Python内置的list、dict、int、str等类型默认不支持弱引用（通过子类化可以）：

class MyList(list):
pass

lst = MyList([1, 2, 3])
ref = weakref.ref(lst) # OK

自定义类默认支持弱引用（包含__weakref__预留槽位），但定义了__slots__后需要显式加入'__weakref__'。

弱引用在框架设计中的实际应用案例——信号系统：

class Signal:
def __init__(self):
self._receivers = {}

def connect(self, receiver, sender=None):
key = (sender, id(receiver))
self._receivers[key] = weakref.ref(receiver)

def send(self, sender=None, **kwargs):
for (s, _), ref in list(self._receivers.items()):
if s is sender:
receiver = ref()
if receiver is not None:
receiver(**kwargs)
else:
# 清理已回收的接收者
del self._receivers[(s, _)]

signal = Signal()
class Subscriber:
def __call__(self, **data):
print(f"收到信号: {data}")

sub = Subscriber()
signal.connect(sub)
signal.send(sender=None, message="hello")
del sub
gc.collect()
signal.send(sender=None, message="world") # 不会调用已回收的对象

这个信号模式的优点是订阅者的生命周期不受信号系统的影响。即使忘记取消订阅，也不会造成内存泄漏。