观察者模式Python实现示例

本文深入剖析了Python中观察者模式在真实项目中的关键实践难点与高阶解决方案：如何用WeakSet自动管理观察者生命周期以杜绝内存泄漏，为何notify时必须遍历列表副本以避免迭代崩溃，怎样智能区分并调度协程与普通observer以保障asyncio事件循环不被阻塞，以及如何通过统一update接口、弱引用注册、显式detach等手段彻底规避循环引用和运行时修改容器等致命陷阱——这些看似琐碎的细节，恰恰决定了观察者模式是从教科书走向健壮生产环境的分水岭。

Observer 类怎么设计才支持动态增删

观察者模式的核心是让被观察者（Subject）能灵活管理观察者列表，而不是写死在类里。常见错误是把 observers 设为普通 list，然后在 notify() 里直接遍历并调用方法——这会导致运行时修改列表（比如某个 observer 自己调用 remove_observer()）触发 RuntimeError: list changed size during iteration。

实操建议：

• 用 set 存储 observer 实例（避免重复注册，且删除是 O(1)）
• notify() 前先转成 tuple 或 list 副本：for obs in tuple(self._observers):
• observer 接口统一定义为实现 update(self, *args, **kwargs)，便于传入事件数据

如何避免循环引用导致内存泄漏

Python 的垃圾回收对循环引用不敏感，如果 Subject 持有 observer 引用，而 observer 又强引用了 Subject（比如闭包、实例属性），对象就无法被释放。典型场景是用 lambda 当 observer：subject.attach(lambda: print("done"))，lambda 里若捕获了 subject 实例，就构成闭环。

实操建议：

• observer 注册时用 weakref.ref 包装，notify() 中先 obs_ref() 判断是否还存活
• 避免在 observer 回调中直接访问 subject 的可变属性；改用传参方式传递必要状态
• 提供显式的 detach_all() 方法，在 subject 生命周期结束前清空引用

asyncio 下 notify() 怎么不阻塞事件循环

同步版 observer 模式在 asyncio 环境中会出问题：一个耗时的 update() 调用会卡住整个 event loop。错误做法是直接在 notify() 里用 await obs.update(...)——因为不是所有 observer 都是协程函数，混用会抛 TypeError: object XXX can't be used in 'await' expression。

实操建议：

• 区分 observer 类型：用 inspect.iscoroutinefunction(getattr(obs, 'update', None)) 判断
• 协程 observer 用 asyncio.create_task(obs.update(...)) 调度，普通 observer 仍同步调用
• 不要在 notify() 中 await 所有任务完成——除非业务明确要求串行等待；多数场景应 fire-and-forget

用标准库 weakref.WeakSet 替代手动管理 observer

自己维护弱引用容易漏掉清理逻辑，比如忘记在 observer 对象销毁时从集合中移除。Python 标准库的 WeakSet 能自动处理：当 observer 实例被 gc，它会自动从集合中消失，无需额外 hook。

实操建议：

• 初始化：self._observers = weakref.WeakSet()
• 注册：self._observers.add(observer)（observer 必须是类实例，不能是函数或 lambda）
• 通知：for obs in list(self._observers): # 必须转 list，避免迭代时被自动清理干扰
• 注意：WeakSet 不支持非 hashable 对象，所以 observer 类需实现 __hash__ 和 __eq__，或确保实例可哈希

真正麻烦的从来不是“怎么写一个 observer”，而是“怎么让它在真实项目里不崩、不卡、不占内存”。这些细节没压平之前，模式只是玩具。

好了，本文到此结束，带大家了解了《观察者模式Python实现示例》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多文章知识！