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

观察者模式

定义

类图

应用项目：气象站

要求

实现类图

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

搭建观察者模式框架

class Observer:
    """观察者"""

    @abstractmethod
    def update(self,temp, humidity, pressure):
        """由主题调用更新"""
        pass


class Subject:
    """主题"""

    @abstractmethod
    def register_observer(self, observer: Observer):
        """注册"""
        pass

    @abstractmethod
    def remove_observer(self, observer: Observer):
        """移除"""
        pass

    @abstractmethod
    def notify_observers(self):
        """通知所有观察者"""
        pass


class DisplayElement:
    def __init__(self):
        pass

    @abstractmethod
    def display(self):
        """显示"""
        pass

实现WeatherData

class WeatherData(Subject):
    def __init__(self):
        self._observers: list[Observer] = []
        self._temp: Optional[float] = None
        self._pressure: Optional[float] = None
        self._humidity: Optional[float] = None

    def 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 = temp
        self._humidity = humidity
        self._pressure = pressure
        self.measurements_changed()

实现布告板

class CurrentConditionsDispaly(Observer, DisplayElement):
    def __init__(self):
        self._weather_data = WeatherData()
        self._weather_data.register_observer(self)
        self._temp: Optional[float] = None
        self._pressure: Optional[float] = None
        self._humidity: Optional[float] = None

    def update(self, temp, humidity, pressure):
        self._temp = temp
        self._humidity = humidity
        self._pressure = pressure
        self.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 Optional


class Observer:
    """观察者"""

    @abstractmethod
    def update(self, temp, humidity, pressure):
        """由主题调用更新"""
        pass


class Subject:
    """主题"""

    @abstractmethod
    def register_observer(self, observer: Observer):
        """注册"""
        pass

    @abstractmethod
    def notify_observers(self):
        """通知所有观察者"""
        pass

    @abstractmethod
    def remove_observer(self, observer: Observer):
        """移除"""
        pass


class DisplayElement:

    @abstractmethod
    def display(self):
        """显示"""
        pass


class WeatherData(Subject):
    def __init__(self):
        self._observers: list[Observer] = []
        self._temp: Optional[float] = None
        self._pressure: Optional[float] = None
        self._humidity: Optional[float] = None

    def 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 = temp
        self._humidity = humidity
        self._pressure = pressure
        self.measurements_changed()


class CurrentConditionsDispaly(Observer, DisplayElement):
    def __init__(self, weather_data: WeatherData):
        self._weather_data = weather_data
        self._weather_data.register_observer(self)
        self._temp: Optional[float] = None
        self._pressure: Optional[float] = None
        self._humidity: Optional[float] = None

    def update(self, temp, humidity, pressure):
        self._temp = temp
        self._humidity = humidity
        self._pressure = pressure
        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)
    
"""运行结果：
4481847152
温度：80
湿度：80
气压：2.34

4481847152
温度：90
湿度：70.5
气压：5

4481845424
温度：90
湿度：70.5
气压：5
"""

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

类图

代码实现

from abc import abstractmethod, ABC
from typing import Optional, Union


class Observer(ABC):
    """观察者"""

    @abstractmethod
    def update(self, observerable, arg):
        """由主题调用更新"""
        pass


class Observerable:
    """可观察者"""
    _observers = []
    changed = False

    def 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 = False

    def set_changed(self):
        self.changed = True


class DisplayElement:

    @abstractmethod
    def display(self):
        """显示"""
        pass


class WeatherData(Observerable):
    _temp: float  # python新版本可不指定默认值
    _humidity: float
    _pressure: float

    def measurements_changed(self):
        self.set_changed()
        self.notify_observers()

    def get_temperature(self) -> float:
        return self._temp

    def get_humility(self) -> float:
        return self._humidity

    def get_pressure(self) -> float:
        return self._pressure

    def set_measurements(self, temp: float, humidity: float, pressure: float):
        self._temp = temp
        self._humidity = humidity
        self._pressure = pressure
        self.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 = observerable
        observerable.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)

总结