Python 观察者模式-平芜编程栈

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

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

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

观察者模式的优点

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

典型应用场景

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

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

fromabcimportABC,abstractmethodfromtypingimportList# 观察者接口（Observer）classObserver(ABC):@abstractmethoddefupdate(self,subject):pass# 具体观察者：图表显示classChartDisplay(Observer):defupdate(self,subject):print(f"【图表】股票价格更新：{subject.stock_name}= ${subject.price:.2f}")# 具体观察者：短信通知classSMSAlert(Observer):def__init__(self,phone_number:str):self.phone_number=phone_numberdefupdate(self,subject):ifsubject.price>150:# 价格超过150时才发短信print(f"【短信】发送到{self.phone_number}：{subject.stock_name}已突破 ${subject.price:.2f}!")# 具体观察者：邮件通知classEmailAlert(Observer):def__init__(self,email:str):self.email=emaildefupdate(self,subject):print(f"【邮件】发送到{self.email}：{subject.stock_name}最新价格 ${subject.price:.2f}")# 主题（Subject / Observable）classStock:def__init__(self,name:str,price:float):self.stock_name=name self.price=price self._observers:List[Observer]=[]defattach(self,observer:Observer):ifobservernotinself._observers:self._observers.append(observer)print(f"{observer.__class__.__name__}已订阅{self.stock_name}")defdetach(self,observer:Observer):ifobserverinself._observers:self._observers.remove(observer)print(f"{observer.__class__.__name__}已取消订阅{self.stock_name}")defnotify(self):print(f"通知所有观察者...")forobserverinself._observers:observer.update(self)defset_price(self,new_price:float):ifnew_price!=self.price:self.price=new_priceprint(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实现自动通知：

classObservable:def__init__(self):self._observers=[]defattach(self,observer):self._observers.append(observer)defdetach(self,observer):self._observers.remove(observer)defnotify(self):forobserverinself._observers:observer(self)# 直接传整个对象classStock(Observable):def__init__(self,name,price):super().__init__()self._name=name self._price=price@propertydefprice(self):returnself._price@price.setterdefprice(self,value):self._price=value self.notify()# 自动通知# 使用stock=Stock("TSLA",800)stock.attach(lambdas: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）