冬季促销!
Python 观察者模式讲解和代码示例
观察者是一种行为设计模式， 允许一个对象将其状态的改变通知其他对象
观察者模式提供了一种作用于任何实现了订阅者接口的对象的机制， 可对其事件进行订阅和取消订阅。
复杂度：
流行度：
使用示例： 观察者模式在 Python 代码中很常见， 特别是在 GUI 组件中。 它提供了在不与其他对象所属类耦合的情况下对其事件做出反应的方式。
识别方法： 该模式可以通过将对象存储在列表中的订阅方法， 和对于面向该列表中对象的更新方法的调用来识别。
概念示例
本例说明了观察者设计模式的结构并重点回答了下面的问题：
- 它由哪些类组成？
- 这些类扮演了哪些角色？
- 模式中的各个元素会以何种方式相互关联？
main.py: 概念示例
from __future__ import annotations
from abc import ABC, abstractmethod
from random import randrange
from typing import List
class Subject(ABC):
"""
The Subject interface declares a set of methods for managing subscribers.
"""
@abstractmethod
def attach(self, observer: Observer) -> None:
"""
Attach an observer to the subject.
"""
pass
@abstractmethod
def detach(self, observer: Observer) -> None:
"""
Detach an observer from the subject.
"""
pass
@abstractmethod
def notify(self) -> None:
"""
Notify all observers about an event.
"""
pass
class ConcreteSubject(Subject):
"""
The Subject owns some important state and notifies observers when the state
changes.
"""
_state: int = None
"""
For the sake of simplicity, the Subject's state, essential to all
subscribers, is stored in this variable.
"""
_observers: List[Observer] = []
"""
List of subscribers. In real life, the list of subscribers can be stored
more comprehensively (categorized by event type, etc.).
"""
def attach(self, observer: Observer) -> None:
print("Subject: Attached an observer.")
self._observers.append(observer)
def detach(self, observer: Observer) -> None:
self._observers.remove(observer)
"""
The subscription management methods.
"""
def notify(self) -> None:
"""
Trigger an update in each subscriber.
"""
print("Subject: Notifying observers...")
for observer in self._observers:
observer.update(self)
def some_business_logic(self) -> None:
"""
Usually, the subscription logic is only a fraction of what a Subject can
really do. Subjects commonly hold some important business logic, that
triggers a notification method whenever something important is about to
happen (or after it).
"""
print("\nSubject: I'm doing something important.")
self._state = randrange(0, 10)
print(f"Subject: My state has just changed to: {self._state}")
self.notify()
class Observer(ABC):
"""
The Observer interface declares the update method, used by subjects.
"""
@abstractmethod
def update(self, subject: Subject) -> None:
"""
Receive update from subject.
"""
pass
"""
Concrete Observers react to the updates issued by the Subject they had been
attached to.
"""
class ConcreteObserverA(Observer):
def update(self, subject: Subject) -> None:
if subject._state < 3:
print("ConcreteObserverA: Reacted to the event")
class ConcreteObserverB(Observer):
def update(self, subject: Subject) -> None:
if subject._state == 0 or subject._state >= 2:
print("ConcreteObserverB: Reacted to the event")
if __name__ == "__main__":
# The client code.
subject = ConcreteSubject()
observer_a = ConcreteObserverA()
subject.attach(observer_a)
observer_b = ConcreteObserverB()
subject.attach(observer_b)
subject.some_business_logic()
subject.some_business_logic()
subject.detach(observer_a)
subject.some_business_logic()
Output.txt: 执行结果
Subject: Attached an observer.
Subject: Attached an observer.
Subject: I'm doing something important.
Subject: My state has just changed to: 0
Subject: Notifying observers...
ConcreteObserverA: Reacted to the event
ConcreteObserverB: Reacted to the event
Subject: I'm doing something important.
Subject: My state has just changed to: 5
Subject: Notifying observers...
ConcreteObserverB: Reacted to the event
Subject: I'm doing something important.
Subject: My state has just changed to: 0
Subject: Notifying observers...
ConcreteObserverB: Reacted to the event
