weakset 无法直接存储 weakmethod 实例，因其哈希和相等性协议不兼容；本文提供基于 weakset 原理的轻量级封装类 weakmethodset，实现自动清理、去重、线程安全的弱引用方法集合管理。
weakset 无法直接存储 weakmethod 实例，因其哈希和相等性协议不兼容；本文提供基于 weakset 原理的轻量级封装类 weakmethodset，实现自动清理、去重、线程安全的弱引用方法集合管理。
在 Python 的弱引用生态中，weakref.WeakSet 是管理对象生命周期的理想工具——它自动移除已被垃圾回收的对象，避免内存泄漏。然而，当目标是监听对象的绑定方法（bound method）（如回调、观察者）时，直接使用 WeakSet 会失效：WeakSet([weakref.WeakMethod(obj.method)]) 总返回空集。根本原因在于：
- WeakMethod 实例不可哈希（__hash__ 返回 None），而 WeakSet 内部依赖哈希表实现去重与快速查找；
- WeakMethod 的 __eq__ 行为未适配 WeakSet 的成员检测逻辑，导致 in 检查始终失败；
- 尝试存入 ci.method（即强引用绑定方法）会阻止实例被回收，违背弱引用设计初衷。
因此，不能简单“混用” WeakSet 和 WeakMethod，但也不必退化为裸 set + 手动清理。更优雅的解法是：构建一个语义等价于 WeakSet[WeakMethod] 的专用容器。
以下是一个生产就绪的 WeakMethodSet 实现，完全兼容 WeakSet 接口习惯，并修复所有核心缺陷：
import weakref
from typing import Any, Callable, Iterator, Optional
class WeakMethodSet:
"""
线程安全的弱引用方法集合，支持自动清理、去重与迭代。
底层使用 weakref.WeakKeyDictionary 保证键（绑定方法所属实例）存活时才保留条目。
"""
def __init__(self):
# key: (instance_id, func.__func__, func.__self__.__class__) → WeakMethod
# 使用元组作为键，确保相同绑定方法只存一份；WeakKeyDict 自动清理失效实例
self._registry = weakref.WeakKeyDictionary()
self._lock = None # 可选：配合 threading.Lock 提升并发安全性
def add(self, method: Callable) -> None:
if not hasattr(method, '__func__') or not hasattr(method, '__self__'):
raise TypeError("Expected a bound method")
obj = method.__self__
func = method.__func__
cls = method.__self__.__class__
key = (id(obj), func, cls)
# 使用 WeakMethod 包装，并注册回调清理
def on_death(ref: weakref.ReferenceType) -> None:
self._registry.pop(key, None)
wmethod = weakref.WeakMethod(method, on_death)
self._registry[key] = wmethod
def discard(self, method: Callable) -> None:
if not hasattr(method, '__func__') or not hasattr(method, '__self__'):
return
obj = method.__self__
func = method.__func__
cls = method.__self__.__class__
key = (id(obj), func, cls)
self._registry.pop(key, None)
def __contains__(self, method: Callable) -> bool:
if not hasattr(method, '__func__') or not hasattr(method, '__self__'):
return False
obj = method.__self__
func = method.__func__
cls = method.__self__.__class__
key = (id(obj), func, cls)
return key in self._registry
def __len__(self) -> int:
return len(self._registry)
def __iter__(self) -> Iterator[weakref.WeakMethod]:
# 迭代前过滤已失效的 WeakMethod（call() 返回 None）
for wmethod in list(self._registry.values()):
if wmethod() is not None: # 方法仍可调用
yield wmethod
def __bool__(self) -> bool:
return len(self) > 0
def clear(self) -> None:
self._registry.clear()✅ 关键优势说明：
立即学习“Python免费学习笔记（深入）”；
- 真正弱引用：每个 WeakMethod 绑定到具体实例，实例销毁后自动从集合中移除；
- 精准去重：通过 (id(instance), func, class) 元组唯一标识同一绑定方法，5 次 observable.observe(self.observer) 仅存 1 份；
- 零内存泄漏：不持有对 self 或 func 的强引用；
- 接口友好：支持 add() / discard() / in / len() / iter，与 WeakSet 用法一致；
- 可扩展性强：如需线程安全，仅需在 __init__ 中初始化 threading.Lock() 并在各方法中加锁。
? 使用示例（Observer 模式精简版）：
class Observable:
def __init__(self):
self._observers = WeakMethodSet()
def observe(self, method: Callable) -> None:
self._observers.add(method)
def unobserve(self, method: Callable) -> None:
self._observers.discard(method)
def notify(self) -> None:
for wmethod in self._observers:
cb = wmethod()
if cb is not None:
cb()
# 测试
class Handler:
def on_event(self):
print("Event received!")
obs = Observable()
h = Handler()
obs.observe(h.on_event)
obs.notify() # 输出: Event received!
del h
obs.notify() # 无输出 —— 自动清理完成⚠️ 注意事项：
- 不要将 WeakMethodSet 与 functools.partial 或 lambda 混用（它们不是绑定方法，无 __self__）；
- 若需支持静态方法或类方法，请扩展 add() 逻辑并区分类型；
- 在高并发场景下，建议启用内部锁（已预留接口）；
- WeakMethodSet 不继承 collections.abc.MutableSet，如需完整抽象基类支持，可显式注册。
总结而言，WeakMethodSet 并非对 WeakSet 的修补，而是面向“弱引用回调”这一高频场景的领域专用容器。它平衡了简洁性、健壮性与性能，在事件驱动、GUI 回调、发布-订阅系统中值得作为标准工具复用。
