如何在Python中优雅实现事件驱动架构的Event类派发机制-Python教程-PHP中文网

Event类必须用弱引用防止内存泄漏，subscribe返回唯一token的unsubscribe句柄，fire需逐个try/except隔离异常，并区分同步/异步回调及跨线程调度。

如何在python中优雅实现事件驱动架构的event类派发机制

Event类设计必须支持弱引用回调，否则内存泄漏不可避免

Python中手动维护事件监听器列表时，若直接用普通函数或绑定方法作为回调，会隐式持有对象引用，导致监听对象无法被GC回收。比如 obj.on_click = lambda: print("clicked") 后又没显式移除，obj 就可能一直存活。

解决方案是用 weakref.WeakKeyDictionary 或 weakref.WeakSet 存储监听器，但更稳妥的是统一用 weakref.ref 包装每个回调：

import weakref
<p>class Event:
def <strong>init</strong>(self):
self._handlers = []</p><pre class="brush:php;toolbar:false;">def subscribe(self, handler):
    # 包装为弱引用，支持函数、方法、lambda（需额外处理）
    if hasattr(handler, '__func__') and hasattr(handler, '__self__'):
        # 绑定方法：弱引用实例，保留函数
        wr = weakref.ref(handler.__self__)
        self._handlers.append((wr, handler.__func__))
    else:
        # 普通函数或lambda，直接弱引用（lambda无法弱引用，此处退化为强引用，但至少不卡死实例）
        self._handlers.append(weakref.ref(handler))

def fire(self, *args, **kwargs):
    # 遍历时需拷贝，避免迭代中修改列表
    for ref in self._handlers[:]:
        handler = ref() if callable(ref) else ref[1](ref[0]()) if ref[0]() else None
        if handler is not None:
            try:
                handler(*args, **kwargs)
            except Exception:
                pass  # 或记录日志，不中断其他监听器

subscribe / unsubscribe 必须成对出现，且 unsubscribe 要能精准匹配原注册项

常见错误是用 lambda 多次订阅后，试图用另一个 lambda 去取消——它们不是同一对象，list.remove() 失败，导致“假性泄漏”。

正确做法是让 subscribe() 返回一个可调用的 unsubscribe 句柄，内部绑定移除逻辑：

返回一个闭包，捕获 handler 引用和 handlers 列表索引（或 id）
避免依赖 list.index()，因为 handler 可能重复或不可哈希
推荐用唯一 token（如 uuid.uuid4()）标记每次订阅，存入字典，unsubscribe 时按 token 删除

示例关键逻辑：

import uuid
<p>def subscribe(self, handler):
token = uuid.uuid4()
self._handlers[token] = weakref.ref(handler)
return lambda: self._handlers.pop(token, None)
</p>

fire() 中异常不应中断后续监听器执行

事件派发本质是“尽力而为”，一个监听器崩溃不该让其余监听器失效。Python 默认 for 循环遇到异常就跳出，这是反模式。

python全能编程助手

SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的双功能合体技能，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途：；核心要点📝 题解模式：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式：输入 Python 代码，自动添加详细中。它将相关步骤、

下载

必须把每个 handler 调用包裹在独立 try/except 中，并区分处理：

静默忽略（适合 UI 事件、日志类回调）
记录到 logging.getLogger(__name__).debug(...)（推荐）
不建议抛出新异常或 re-raise，除非上层明确要求“阻塞式事件”

还要注意：不要在 fire() 内部做耗时操作（如网络请求、文件读写），这会卡住整个事件循环；应交由线程池或 asyncio.create_task() 转移。

async def 回调需要单独支持，不能和同步混用

如果项目已用 asyncio，但 Event.fire() 是同步方法，直接 await 一个协程会导致 RuntimeWarning: coroutine 'xxx' was never awaited，甚至静默失败。

有两种实用路径：

提供分离接口：fire_sync() 和 fire_async()，后者返回 awaitable 或调度到事件循环
自动检测回调类型：用 inspect.iscoroutinefunction(handler) 或 asyncio.iscoroutine(handler(...)) 分流

但注意：自动检测有性能开销，且无法识别装饰器包装后的协程；生产环境建议显式区分，避免魔改带来的调试困难。

真正棘手的是跨线程调用 —— 比如主线程 fire，但某个 handler 必须在 asyncio loop 线程执行。这时得用 asyncio.run_coroutine_threadsafe()，而不是简单 await。