Python 观察者模式

Python 中的观察者模式（Observer Pattern）

观察者模式是一种行为型设计模式，其核心目的是：
定义对象间的一种一对多依赖关系，当一个对象（主题/被观察者）的状态发生改变时，所有依赖它的对象（观察者）都得到通知并自动更新。

形象比喻：就像订阅报纸——你（观察者）订阅了报纸（主题），每当有新报纸出版，邮递员就会自动送到你家。

观察者模式的优点

• 实现主题和观察者之间的松耦合（主题只知道观察者列表，不关心具体类型）
• 支持广播通信（一个变化通知多个对象）
• 符合“开闭原则”：新增观察者无需修改主题代码

典型应用场景

• GUI 系统：按钮点击通知多个监听器
• 事件驱动编程（如 tkinter、PyQt 的信号槽）
• 发布-订阅系统（消息队列、事件总线）
• 数据变更通知（如股票价格更新、天气预报）
• MVC 架构中的 View 更新

Python 实现示例：股票价格监控

from abc import ABC, abstractmethod
from typing import List

# 观察者接口（Observer）
class Observer(ABC):
    @abstractmethod
    def update(self, subject):
        pass

# 具体观察者：图表显示
class ChartDisplay(Observer):
    def update(self, subject):
        print(f"【图表】股票价格更新：{subject.stock_name} = ${subject.price:.2f}")

# 具体观察者：短信通知
class SMSAlert(Observer):
    def __init__(self, phone_number: str):
        self.phone_number = phone_number

    def update(self, subject):
        if subject.price > 150:  # 价格超过150时才发短信
            print(f"【短信】发送到 {self.phone_number}：{subject.stock_name} 已突破 ${subject.price:.2f}!")

# 具体观察者：邮件通知
class EmailAlert(Observer):
    def __init__(self, email: str):
        self.email = email

    def update(self, subject):
        print(f"【邮件】发送到 {self.email}：{subject.stock_name} 最新价格 ${subject.price:.2f}")

# 主题（Subject / Observable）
class Stock:
    def __init__(self, name: str, price: float):
        self.stock_name = name
        self.price = price
        self._observers: List[Observer] = []

    def attach(self, observer: Observer):
        if observer not in self._observers:
            self._observers.append(observer)
            print(f"{observer.__class__.__name__} 已订阅 {self.stock_name}")

    def detach(self, observer: Observer):
        if observer in self._observers:
            self._observers.remove(observer)
            print(f"{observer.__class__.__name__} 已取消订阅 {self.stock_name}")

    def notify(self):
        print(f"通知所有观察者...")
        for observer in self._observers:
            observer.update(self)

    def set_price(self, new_price: float):
        if new_price != self.price:
            self.price = new_price
            print(f"{self.stock_name} 价格变动 → ${new_price:.2f}")
            self.notify()

# 客户端使用
if __name__ == "__main__":
    # 创建股票（主题）
    apple_stock = Stock("AAPL", 145.0)

    # 创建观察者
    chart = ChartDisplay()
    sms = SMSAlert("138-1234-5678")
    email = EmailAlert("investor@example.com")

    # 订阅
    apple_stock.attach(chart)
    apple_stock.attach(sms)
    apple_stock.attach(email)

    print("\n--- 第一次价格更新 ---")
    apple_stock.set_price(152.0)  # 触发通知

    print("\n--- 第二次价格更新 ---")
    apple_stock.set_price(148.0)  # 短信不触发（未超150）

    print("\n--- 取消短信订阅 ---")
    apple_stock.detach(sms)

    print("\n--- 第三次价格更新 ---")
    apple_stock.set_price(160.0)  # 只有图表和邮件收到通知

输出：

ChartDisplay 已订阅 AAPL
SMSAlert 已订阅 AAPL
EmailAlert 已订阅 AAPL

--- 第一次价格更新 ---
AAPL 价格变动 → $152.00
通知所有观察者...
【图表】股票价格更新：AAPL = $152.00
【短信】发送到 138-1234-5678：AAPL 已突破 $152.00!
【邮件】发送到 investor@example.com：AAPL 最新价格 $152.00

--- 第二次价格更新 ---
AAPL 价格变动 → $148.00
通知所有观察者...
【图表】股票价格更新：AAPL = $148.00
【邮件】发送到 investor@example.com：AAPL 最新价格 $148.00

--- 取消短信订阅 ---
SMSAlert 已取消订阅 AAPL

--- 第三次价格更新 ---
AAPL 价格变动 → $160.00
通知所有观察者...
【图表】股票价格更新：AAPL = $160.00
【邮件】发送到 investor@example.com：AAPL 最新价格 $160.00

Pythonic 更简洁实现：使用属性装饰器

Python 中可以用 @property 实现自动通知：

class Observable:
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    def notify(self):
        for observer in self._observers:
            observer(self)  # 直接传整个对象

class Stock(Observable):
    def __init__(self, name, price):
        super().__init__()
        self._name = name
        self._price = price

    @property
    def price(self):
        return self._price

    @price.setter
    def price(self, value):
        self._price = value
        self.notify()  # 自动通知

# 使用
stock = Stock("TSLA", 800)
stock.attach(lambda s: print(f"价格更新: {s._name} = ${s._price}"))
stock.price = 850  # 自动触发

内置支持：事件库

Python 标准库没有内置观察者，但第三方库很成熟：

• blinker：轻量级信号系统
• PyDispatcher：经典事件分发
• GUI 框架（如 tkinter、PyQt、PySide）自带信号槽机制

观察者模式结构总结

角色	说明
Subject	主题，维护观察者列表，提供 attach/detach/notify
Observer	抽象观察者，定义 update 接口
ConcreteSubject	具体主题（如 Stock），状态变化时调用 notify
ConcreteObserver	具体观察者（如 ChartDisplay、SMSAlert）

观察者模式 vs 发布-订阅模式

• 观察者模式：主题直接持有观察者引用（紧耦合于同一进程）
• 发布-订阅（Pub/Sub）：通过消息中间件解耦（支持分布式，如 Redis、Kafka）

Python 中的实用建议

• 小型项目：手写观察者（如上例）
• 中型项目：使用 @property + 回调列表
• GUI/事件驱动：直接用框架的信号机制
• 大型/分布式：使用消息队列（Celery、RabbitMQ）

注意事项

• 通知顺序不确定（除非显式排序）
• 避免在 update 中修改主题状态（可能无限循环）
• 多线程时需加锁保护观察者列表
• 取消订阅很重要（避免内存泄漏）

观察者模式是事件驱动编程的核心，在 Python 的 GUI、Web 框架、异步系统中无处不在。

如果你想看更高级的例子（如多线程安全观察者、结合 asyncio 的异步通知、使用 blinker 库的信号系统），或者与其他模式结合（如观察者 + 策略），随时告诉我！