observer类应设计为可调用对象(支持__call__),用weakset存储以自动清理销毁实例,注册时检查hasattr(obj, '__call__'),通知时用try/except隔离异常并支持**kwargs传参,配合protocol做类型提示。

Observer 类怎么设计才支持动态增删监听器
观察者模式的核心是让被观察者(Subject)不依赖具体监听逻辑,只维护一组可调用对象。用 list 存储观察者最直接,但必须确保添加的是可调用对象,且避免重复注册或已销毁对象引发的 AttributeError。
- 用
weakref.WeakSet替代普通list,能自动清理被垃圾回收的观察者实例,防止内存泄漏 - 注册时检查
hasattr(observer, '__call__'),避免传入非函数/方法导致运行时报错 - 移除观察者建议用
remove()而非pop(),否则容易因索引错位漏删;若不确定是否已注册,用discard()(配合WeakSet)更安全
notify() 方法里为什么不能直接 for 循环调用观察者
直接遍历列表并同步调用每个观察者,会阻塞主线程、放大异常影响——任一观察者抛出未捕获异常,后续监听器全部跳过。这不是“事件驱动”,而是“串行回调”。
- 对每个观察者调用包一层
try...except Exception,单独捕获并记录错误,不影响其余执行 - 如需异步通知,用
threading.Thread启动新线程,或交给asyncio.create_task()(需整个系统协程化) - 注意:Python 的 GIL 会让多线程在 CPU 密集型任务中收益有限,I/O 型事件才适合线程分发
如何让事件携带不同数据而不破坏 Observer 接口一致性
硬编码参数列表(如 update(self, data1, data2))会导致新增事件类型就得改所有观察者实现。应该统一用关键字参数传递上下文。
- 被观察者调用
observer(**event_data),其中event_data是 dict,例如{'type': 'user_login', 'user_id': 1024, 'ip': '192.168.1.5'} - 观察者方法签名保持为
def on_event(self, **kwargs):,按需取键,不关心多余字段 - 避免用位置参数传 event,否则增加新字段要同步改所有调用点和接收方,违反开闭原则
Python 3.12+ 中用 Protocol 实现类型提示有什么实际好处
没有类型提示时,IDE 不知道 observer 应该长什么样,重构易出错;运行期也缺乏校验。用 Protocol 定义结构契约,比抽象基类更轻量、更符合鸭子类型。
from typing import Protocol
class EventObserver(Protocol):
def __call__(self, *, type: str, **kwargs) -> None: ...
- 静态检查工具(如 mypy)能验证传入对象是否满足协议,提前发现
observer缺少__call__或签名不符 - 不强制继承,普通函数、绑定方法、带
__call__的类实例都可直接用 - 注意:Protocol 是运行期擦除的,仅用于类型检查,不会影响性能或行为
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











