当前位置：首页 > news >正文

Python观察者模式详解：从理论到实战

news 2025/7/20 8:38:09

一、模式简介

观察者模式（Observer Pattern）是一种行为设计模式，允许对象（观察者）订阅另一个对象（被观察者）的状态变化，并在状态改变时自动接收通知。这种模式完美解决了"一对多"的对象间通信问题。

核心思想

发布-订阅机制：被观察者维护观察者列表，当状态变化时自动通知所有观察者
松耦合设计：被观察者不需要知道观察者的具体实现，只需通过统一接口通信
动态关系：观察者可以随时订阅或取消订阅

二、模式组成

# 抽象观察者接口
class Observer:def update(self, message):"""接收通知的接口方法"""pass# 抽象被观察者接口
class Subject:def __init__(self):self._observers = []  # 观察者列表def attach(self, observer):"""添加观察者"""if observer not in self._observers:self._observers.append(observer)def detach(self, observer):"""移除观察者"""try:self._observers.remove(observer)except ValueError:passdef notify(self, message):"""通知所有观察者"""for observer in self._observers:observer.update(message)

三、实战案例：天气预报系统

场景描述

天气站（被观察者）实时监测天气数据
手机APP、显示屏（观察者）需要实时接收天气更新

完整实现

# 具体被观察者：天气站
class WeatherStation(Subject):def __init__(self):super().__init__()self._temperature = 0@propertydef temperature(self):return self._temperature@temperature.setterdef temperature(self, value):self._temperature = valueself.notify(f"当前温度更新为：{value}℃")  # 触发通知# 具体观察者：手机APP
class MobileApp(Observer):def update(self, message):print(f"手机APP收到通知：{message}")# 具体观察者：电子显示屏
class DisplayBoard(Observer):def update(self, message):print(f"显示屏更新：{message}")# 使用示例
if __name__ == "__main__":# 创建被观察者weather_station = WeatherStation()# 创建观察者app = MobileApp()display = DisplayBoard()# 订阅服务weather_station.attach(app)weather_station.attach(display)# 温度更新（自动触发通知）weather_station.temperature = 25weather_station.temperature = 28# 取消订阅weather_station.detach(app)weather_station.temperature = 30  # 只有显示屏会收到通知

输出结果

手机APP收到通知：当前温度更新为：25℃
显示屏更新：当前温度更新为：25℃
手机APP收到通知：当前温度更新为：28℃
显示屏更新：当前温度更新为：28℃
显示屏更新：当前温度更新为：30℃

四、模式优势

松耦合设计：被观察者无需知道观察者的具体实现
动态关系：运行时可以自由添加/移除观察者
开闭原则：新增观察者无需修改被观察者代码
广播通信：支持一对多的通知机制

五、应用场景

GUI事件处理（如按钮点击通知多个组件）
消息订阅系统（如新闻推送）
分布式系统通信（如微服务间的事件通知）
数据监控场景（如股票价格实时更新）
游戏开发（如玩家状态变化通知）

六、进阶技巧

1. 使用弱引用避免内存泄漏

import weakrefclass Subject:def __init__(self):self._observers = weakref.WeakSet()  # 使用弱引用集合

2. 带过滤器的通知

def notify(self, message, priority="normal"):for observer in self._observers:if hasattr(observer, 'priority_filter'):if observer.priority_filter(priority):observer.update(message)else:observer.update(message)

3. 异步通知（使用线程池）

from concurrent.futures import ThreadPoolExecutorclass AsyncSubject(Subject):def __init__(self):super().__init__()self._executor = ThreadPoolExecutor(max_workers=5)def notify(self, message):for observer in self._observers:self._executor.submit(observer.update, message)

七、模式对比

特性	观察者模式	发布-订阅模式
耦合度	中等（直接依赖）	松散（通过中间件）
通信方式	直接通知	通过消息通道
适用场景	单一系统内	跨系统/微服务架构
实现复杂度	简单	较高