基于Python学习《Head First设计模式》第二章 观察者模式

观察者模式

定义

类图

应用项目：气象站

要求

实现类图

实现方式1-更新所有观察者的所有数据

搭建观察者模式框架

class Observer:"""观察者"""@abstractmethoddef update(self,temp, humidity, pressure):"""由主题调用更新"""passclass Subject:"""主题"""@abstractmethoddef register_observer(self, observer: Observer):"""注册"""pass@abstractmethoddef remove_observer(self, observer: Observer):"""移除"""pass@abstractmethoddef notify_observers(self):"""通知所有观察者"""passclass DisplayElement:def __init__(self):pass@abstractmethoddef display(self):"""显示"""pass

实现WeatherData

class WeatherData(Subject):def __init__(self):self._observers: list[Observer] = []self._temp: Optional[float] = Noneself._pressure: Optional[float] = Noneself._humidity: Optional[float] = Nonedef register_observer(self, observer: Observer):self._observers.append(observer)def remove_observer(self, observer: Observer):self._observers.remove(observer)def notify_observers(self):for obs in self._observers:obs.update(self._temp, self._humidity, self._pressure)def measurements_changed(self):self.notify_observers()def set_measurements(self, temp: float, humidity: float, pressure: float):self._temp = tempself._humidity = humidityself._pressure = pressureself.measurements_changed()

实现布告板

class CurrentConditionsDispaly(Observer, DisplayElement):def __init__(self):self._weather_data = WeatherData()self._weather_data.register_observer(self)self._temp: Optional[float] = Noneself._pressure: Optional[float] = Noneself._humidity: Optional[float] = Nonedef update(self, temp, humidity, pressure):self._temp = tempself._humidity = humidityself._pressure = pressureself.display()def display(self):print(id(self))  # 打印当前示例id，用于区分print(f"温度：{self._temp}"f"湿度：{self._humidity}"f"气压：{self._pressure}")

测试&运行

if __name__ == '__main__':weather = WeatherData()display = CurrentConditionsDispaly(weather)weather.set_measurements(80, 80, 2.34)display2 = CurrentConditionsDispaly(weather)weather.set_measurements(90, 70.5, 5)

完整代码

from abc import abstractmethod
from typing import Optionalclass Observer:"""观察者"""@abstractmethoddef update(self, temp, humidity, pressure):"""由主题调用更新"""passclass Subject:"""主题"""@abstractmethoddef register_observer(self, observer: Observer):"""注册"""pass@abstractmethoddef notify_observers(self):"""通知所有观察者"""pass@abstractmethoddef remove_observer(self, observer: Observer):"""移除"""passclass DisplayElement:@abstractmethoddef display(self):"""显示"""passclass WeatherData(Subject):def __init__(self):self._observers: list[Observer] = []self._temp: Optional[float] = Noneself._pressure: Optional[float] = Noneself._humidity: Optional[float] = Nonedef register_observer(self, observer: Observer):self._observers.append(observer)def remove_observer(self, observer: Observer):self._observers.remove(observer)def notify_observers(self):for obs in self._observers:obs.update(self._temp, self._humidity, self._pressure)def measurements_changed(self):self.notify_observers()def set_measurements(self, temp: float, humidity: float, pressure: float):self._temp = tempself._humidity = humidityself._pressure = pressureself.measurements_changed()class CurrentConditionsDispaly(Observer, DisplayElement):def __init__(self, weather_data: WeatherData):self._weather_data = weather_dataself._weather_data.register_observer(self)self._temp: Optional[float] = Noneself._pressure: Optional[float] = Noneself._humidity: Optional[float] = Nonedef update(self, temp, humidity, pressure):self._temp = tempself._humidity = humidityself._pressure = pressureself.display()def display(self):print(id(self))print(f"温度：{self._temp}\n"f"湿度：{self._humidity}\n"f"气压：{self._pressure}\n")if __name__ == '__main__':weather = WeatherData()display = CurrentConditionsDispaly(weather)weather.set_measurements(80, 80, 2.34)display2 = CurrentConditionsDispaly(weather)weather.set_measurements(90, 70.5, 5)"""运行结果：
4481847152
温度：80
湿度：80
气压：2.344481847152
温度：90
湿度：70.5
气压：54481845424
温度：90
湿度：70.5
气压：5
"""

实现方式2-通知观察者自行拉取数据更新

类图

代码实现

from abc import abstractmethod, ABC
from typing import Optional, Unionclass Observer(ABC):"""观察者"""@abstractmethoddef update(self, observerable, arg):"""由主题调用更新"""passclass Observerable:"""可观察者"""_observers = []changed = Falsedef register_observer(self, observer: Observer):self._observers.append(observer)def remove_observer(self, observer: Observer):self._observers.remove(observer)def notify_observers(self, arg=None):if self.changed:for obs in self._observers:obs.update(self, arg)else:self.changed = Falsedef set_changed(self):self.changed = Trueclass DisplayElement:@abstractmethoddef display(self):"""显示"""passclass WeatherData(Observerable):_temp: float  # python新版本可不指定默认值_humidity: float_pressure: floatdef measurements_changed(self):self.set_changed()self.notify_observers()def get_temperature(self) -> float:return self._tempdef get_humility(self) -> float:return self._humiditydef get_pressure(self) -> float:return self._pressuredef set_measurements(self, temp: float, humidity: float, pressure: float):self._temp = tempself._humidity = humidityself._pressure = pressureself.measurements_changed()class CurrentConditionsDispaly(Observer, DisplayElement):_temp: float  # python新版本可不指定默认值_humidity: Optional[float]  # 值类型可以为float和None时，可以用 Optional[float] 相当于 Union[float, None]_pressure: Union[float, None]def __init__(self, observerable: Observerable):self._observerable = observerableobserverable.register_observer(self)def update(self, observerable, arg):if isinstance(observerable, WeatherData):self._temp = observerable.get_temperature()self._pressure = observerable.get_pressure()self._humidity = observerable.get_humility()self.display()def display(self):print(id(self))print(f"温度：{self._temp}\n"f"湿度：{self._humidity}\n"f"气压：{self._pressure}\n")if __name__ == '__main__':weather = WeatherData()display = CurrentConditionsDispaly(weather)weather.set_measurements(80, 80, 2.34)display2 = CurrentConditionsDispaly(weather)weather.set_measurements(90, 70.5, 5)

总结