Наблюдатель на Python
Наблюдатель — это поведенческий паттерн, который позволяет объектам оповещать другие объекты об изменениях своего состояния.
При этом наблюдатели могут свободно подписываться и отписываться от этих оповещений.
Сложность:
Популярность:
Применимость: Наблюдатель можно часто встретить в Python коде, особенно там, где применяется событийная модель отношений между компонентами. Наблюдатель позволяет отдельным компонентам реагировать на события, происходящие в других компонентах.
Признаки применения паттерна: Наблюдатель можно определить по механизму подписки и методам оповещения, которые вызывают компоненты программы.
Концептуальный пример
Этот пример показывает структуру паттерна Наблюдатель, а именно — из каких классов он состоит, какие роли эти классы выполняют и как они взаимодействуют друг с другом.
main.py: Пример структуры паттерна
from __future__ import annotations
from abc import ABC, abstractmethod
from random import randrange
from typing import List
class Subject(ABC):
"""
Интерфейс издателя объявляет набор методов для управлениями подписчиками.
"""
@abstractmethod
def attach(self, observer: Observer) -> None:
"""
Присоединяет наблюдателя к издателю.
"""
pass
@abstractmethod
def detach(self, observer: Observer) -> None:
"""
Отсоединяет наблюдателя от издателя.
"""
pass
@abstractmethod
def notify(self) -> None:
"""
Уведомляет всех наблюдателей о событии.
"""
pass
class ConcreteSubject(Subject):
"""
Издатель владеет некоторым важным состоянием и оповещает наблюдателей о его
изменениях.
"""
_state: int = None
"""
Для удобства в этой переменной хранится состояние Издателя, необходимое всем
подписчикам.
"""
_observers: List[Observer] = []
"""
Список подписчиков. В реальной жизни список подписчиков может храниться в
более подробном виде (классифицируется по типу события и т.д.)
"""
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)
"""
Методы управления подпиской.
"""
def notify(self) -> None:
"""
Запуск обновления в каждом подписчике.
"""
print("Subject: Notifying observers...")
for observer in self._observers:
observer.update(self)
def some_business_logic(self) -> None:
"""
Обычно логика подписки – только часть того, что делает Издатель.
Издатели часто содержат некоторую важную бизнес-логику, которая
запускает метод уведомления всякий раз, когда должно произойти что-то
важное (или после этого).
"""
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):
"""
Интерфейс Наблюдателя объявляет метод уведомления, который издатели
используют для оповещения своих подписчиков.
"""
@abstractmethod
def update(self, subject: Subject) -> None:
"""
Получить обновление от субъекта.
"""
pass
"""
Конкретные Наблюдатели реагируют на обновления, выпущенные Издателем, к которому
они прикреплены.
"""
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__":
# Клиентский код.
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
