16、观察者模式实现（一）：基于Python内置的Observer模式（使用列表管理观察者）

观察者模式，说白了就是「发布-订阅」那一套。一个对象状态变了，其他一堆对象要跟着更新。我最早接触这个模式是在做实时监控系统的时候，那时候还不知道这叫观察者模式，只是觉得「数据变了，界面得刷新」这个需求太常见了。

今天咱们先聊最朴素的实现方式——用列表来管理观察者。Python 内置的机制其实已经帮我们铺好了路，你甚至不需要额外装任何库。

观察者模式的核心三要素

要理解这个模式，你记住三个角色就够了：

  • 主题（Subject）：被观察的对象，它持有观察者列表，负责通知
  • 观察者（Observer）：订阅主题的对象，收到通知后执行自己的逻辑
  • 通知机制：主题遍历观察者列表，逐个调用更新方法

嗯，这里要注意：观察者模式不是「推」就是「拉」。推模式是主题把数据直接传给观察者，拉模式是观察者自己来取。我个人习惯用推模式，因为接口更清晰，观察者不需要知道主题的内部结构。

用列表管理观察者：最直观的实现

Python 里实现这个太简单了。一个列表，几个方法，搞定。我在项目中遇到过好几次这样的场景：用户登录状态变了，多个模块（导航栏、权限菜单、个人中心）都要刷新。用观察者模式，代码瞬间清爽。

来看代码：

class Subject:
    """主题类，管理观察者列表"""
    def __init__(self):
        self._observers = []  # 观察者列表
    
    def attach(self, observer):
        """注册观察者"""
        if observer not in self._observers:
            self._observers.append(observer)
            print(f"观察者 {observer.name} 已注册")
    
    def detach(self, observer):
        """注销观察者"""
        try:
            self._observers.remove(observer)
            print(f"观察者 {observer.name} 已注销")
        except ValueError:
            print("观察者不存在")
    
    def notify(self, message=None):
        """通知所有观察者"""
        print(f"主题发出通知: {message}")
        for observer in self._observers:
            observer.update(message)


class Observer:
    """观察者基类"""
    def __init__(self, name):
        self.name = name
    
    def update(self, message):
        """收到通知后的更新方法，子类重写"""
        raise NotImplementedError("子类必须实现 update 方法")


class UserObserver(Observer):
    """具体的观察者：用户模块"""
    def update(self, message):
        print(f"[{self.name}] 收到通知: {message}，正在刷新用户信息...")


class MenuObserver(Observer):
    """具体的观察者：菜单模块"""
    def update(self, message):
        print(f"[{self.name}] 收到通知: {message}，正在更新菜单权限...")


class LogObserver(Observer):
    """具体的观察者：日志模块"""
    def update(self, message):
        print(f"[{self.name}] 收到通知: {message}，正在记录日志...")

你看，代码量不大。主题只管维护列表和遍历通知，观察者只管实现自己的 update 逻辑。这就是「开闭原则」的体现——新增一个观察者，不需要改主题的代码。

运行一下看看效果

if __name__ == "__main__":
    # 创建主题
    subject = Subject()
    
    # 创建观察者
    user_obs = UserObserver("用户模块")
    menu_obs = MenuObserver("菜单模块")
    log_obs = LogObserver("日志模块")
    
    # 注册观察者
    subject.attach(user_obs)
    subject.attach(menu_obs)
    subject.attach(log_obs)
    
    # 触发通知
    subject.notify("用户登录成功")
    
    print("\n--- 注销菜单观察者 ---")
    subject.detach(menu_obs)
    
    subject.notify("用户退出登录")

输出结果：

观察者 用户模块 已注册
观察者 菜单模块 已注册
观察者 日志模块 已注册
主题发出通知: 用户登录成功
[用户模块] 收到通知: 用户登录成功，正在刷新用户信息...
[菜单模块] 收到通知: 用户登录成功，正在更新菜单权限...
[日志模块] 收到通知: 用户登录成功，正在记录日志...

--- 注销菜单观察者 ---
观察者 菜单模块 已注销
主题发出通知: 用户退出登录
[用户模块] 收到通知: 用户退出登录，正在刷新用户信息...
[日志模块] 收到通知: 用户退出登录，正在记录日志...

为什么会这样？因为列表里少了一个观察者，通知时自然就跳过了。你想想看，这种动态增删观察者的能力，在实际系统中多有用——比如用户关闭了某个面板，就不用再通知它了。

这个实现有什么坑？

⚠️ 我曾经踩过的坑：
  • 观察者列表的线程安全问题：如果在多线程环境下遍历列表时，另一个线程删除了观察者，会抛出 RuntimeError。解决方案是用锁或者拷贝列表。
  • 循环通知：观察者的 update 方法里又去修改主题，导致无限递归。我建议在通知时加一个递归深度限制。
  • 内存泄漏：观察者注册后忘记注销，主题持有对观察者的引用，导致 GC 无法回收。尤其是用 lambda 或闭包做观察者时，更要小心。
💡 我的小技巧： 如果你只是做简单的通知，可以用 Python 的 weakref 来管理观察者列表，这样观察者被销毁后自动从列表中移除，省去手动 detach 的麻烦。

用 SVG 画一张结构图

下面这张图展示了观察者模式的核心流程，我建议你多看几遍，理解「注册-通知-更新」这个闭环：

主题 (Subject) 观察者列表 [_obs] 观察者A (update) 观察者B (update) 观察者C (update) 遍历通知 attach / detach 核心流程：注册观察者 → 状态变化 → 遍历列表 → 逐个更新 主题 观察者 通知流 注册/注销

什么时候用这种实现？

说实话，用列表管理观察者是最基础的方式，适合以下场景：

  • 观察者数量不多（几十个以内）
  • 通知频率不高（不是每毫秒都在通知）
  • 不需要复杂的过滤或优先级排序
  • 你只是想快速实现一个发布-订阅机制

我在做内部工具的时候经常用这种方式，简单直接，出了问题也好排查。但如果是高并发、高频率的场景，我会考虑用事件队列或者消息中间件，这个后面再聊。

总结一下

用 Python 列表实现观察者模式，核心就三件事：

  1. 主题维护一个 _observers 列表
  2. 提供 attach 和 detach 方法管理观察者
  3. 通知时遍历列表，调用每个观察者的 update

说白了，这就是一个「回调函数列表」的面向对象版本。你想想看，是不是这个理？

📌 记住这个原则：观察者模式的核心是「解耦」。主题不需要知道观察者是谁，观察者也不需要知道主题内部怎么实现。它们只通过一个 update 接口通信。

下一节我们会在这个基础上，引入 Python 的 weakref 来解决内存泄漏问题，让实现更健壮。但今天这个版本，已经足够你应付大部分日常需求了。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →