用 weakref.WeakSet 实现事件分发器更轻量稳定，避免内存泄漏；支持按事件名路由、异常隔离、异步协程与同步函数混合调用，并兼顾线程安全扩展性。
直接用标准库的 weakref + 自定义事件分发器，比手写 Observer 类更轻、更稳、更少内存泄漏风险。
为什么不用继承式 Observer 类
很多教程教你写一个 Observer 抽象基类和 Subject 管理列表，但实际项目里容易出三类问题：订阅者被强引用导致无法 GC、重复订阅难去重、通知时抛异常中断后续监听。Python 的动态性让它更适合用“注册回调函数”而非“强制实现接口”的方式。
- 所有监听函数应以弱引用方式存储，否则 GUI 控件、Flask 请求上下文等短生命周期对象会拖垮内存
- 不依赖
abc.ABC，允许普通函数、lambda、绑定方法甚至functools.partial直接注册 - 通知过程需捕获单个监听器异常，不影响其余监听器执行
用 weakref.WeakSet 管理监听器
核心是避免循环引用：Subject 持有监听器引用 → 监听器（如类实例）又持有 Subject 引用 → 两者都无法被回收。用 WeakSet 可自动清理已销毁的对象。
import weakref
<p>class EventBroker:
def <strong>init</strong>(self):
self._listeners = weakref.WeakSet()</p><pre class="brush:php;toolbar:false;">def subscribe(self, callback):
self._listeners.add(callback)
def notify(self, *args, **kwargs):
# 遍历时转成 list，防止迭代中集合被弱引用自动清理导致 RuntimeError
for cb in list(self._listeners):
try:
cb(*args, **kwargs)
except Exception:
pass # 或记录日志，但不停止后续通知
注意：WeakSet 只支持 hashable 对象，所以不能直接存未绑定的实例方法（如 obj.on_event）。解决办法是用 weakref.WeakKeyDictionary 或封装一层——但更简单的是要求业务方显式传函数，或用 functools.partial 绑定。
按事件类型分发（类似 EventEmitter）
真实业务不会只有一种事件。硬编码一个 notify() 不够用，需要按字符串事件名路由：
class EventBus:
def __init__(self):
self._handlers = {}
<pre class="brush:php;toolbar:false;">def subscribe(self, event_name, callback):
if event_name not in self._handlers:
self._handlers[event_name] = weakref.WeakSet()
self._handlers[event_name].add(callback)
def emit(self, event_name, *args, **kwargs):
if event_name not in self._handlers:
return
for cb in list(self._handlers[event_name]):
try:
cb(*args, **kwargs)
except Exception:
pass
- 事件名建议用
"user.created"这种点分命名，方便层级过滤（比如监听"user.*"需额外加匹配逻辑） - 不要在
emit里做深拷贝参数——性能敏感场景下，由监听器自行决定是否复制数据 - 测试时可临时替换
_handlers为普通 dict，绕过弱引用干扰断点调试
与 asyncio 兼容的关键点
如果业务模块用了异步（比如 FastAPI 后端或 aiohttp 客户端），同步通知会阻塞 event loop。此时不能直接调用 await cb()，因为 cb 可能是同步函数。
- 区分协程函数和普通函数：
inspect.iscoroutinefunction(cb) - 在
asyncio.get_event_loop().create_task(...)中调度协程，普通函数仍用cb() - 避免在
emit中await所有监听器——这会让通知变成串行，失去解耦意义；应并发触发，错误各自处理
真正棘手的是跨线程：主线程注册的监听器，在子线程里 emit 会失效（WeakSet 线程不安全）。这种场景必须换用线程安全的结构（如 threading.RLock + 普通 set），并手动管理生命周期。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
