Паттерн Наблюдатель в Python

Паттерн Наблюдатель — это поведенческий паттерн проектирования, который определяет зависимость «один ко многим», позволяя нескольким объектам-наблюдателям одновременно следить за определённым объектом-субъектом. Когда состояние объекта-субъекта изменяется, все его наблюдатели получают уведомление и автоматически обновляются.

Паттерн Наблюдатель в повседневной жизни

Представьте сценарий подписки на газету:

  • Редакция газеты(Субъект) публикует новости
  • Подписчики(Наблюдатель) подписываются на газету
  • Когда публикуется новая газета, все подписчики автоматически получают газету
  • Подписчики могут отменить свою подписку в любое время

Этот механизм «публикация-подписка» — типичный пример применения паттерна «Наблюдатель».

Основные компоненты паттерна «Наблюдатель»

Паттерн «Наблюдатель» содержит четыре ключевые роли:

1. Субъект

  • Поддерживает список наблюдателей
  • Предоставляет методы для регистрации и отмены регистрации наблюдателей
  • Уведомляет всех наблюдателей об изменениях состояния

2. Конкретный субъект

  • Хранит конкретную информацию о состоянии
  • Уведомляет наблюдателей при изменении состояния

3. Наблюдатель

  • Определяет интерфейс обновления для получения уведомлений от субъекта

4. Конкретный наблюдатель

  • Реализует интерфейс обновления, чтобы оставаться согласованным с состоянием субъекта

Реализация паттерна «Наблюдатель»

Давайте разберём реализацию паттерна «Наблюдатель» на полном примере.

Базовая реализация

Пример

from abc import ABC, abstractmethod
from typing import List

# Интерфейс Observer
class Observer(ABC):
    @abstractmethod
    def update(self, message: str):
        """Метод обновления Observer"""
        pass

# Интерфейс Subject
class Subject(ABC):
    @abstractmethod
    def register_observer(self, observer: Observer):
        """Зарегистрировать наблюдателя"""
        pass
   
    @abstractmethod
    def remove_observer(self, observer: Observer):
        """Удалить наблюдателя"""
        pass
   
    @abstractmethod
    def notify_observers(self):
        """Уведомить всех наблюдателей"""
        pass

# Конкретный субъект - издатель новостей
class NewsPublisher(Subject):
    def __init__(self):
        self._observers: List[Observer] = []
        self._latest_news = ""
   
    def register_observer(self, observer: Observer):
        """Зарегистрировать наблюдателя"""
        if observer not in self_observers:
            self._observers.append(observer)
            print(f"Наблюдатель {observer.__class__.__name__} успешно зарегистрирован")
   
    def remove_observer(self, observer: Observer):
        """Удалить наблюдателя"""
        if observer in self._observers:
            self._observers.remove(observer)
            print(f"Наблюдатель {observer.__class__.__name__} успешно удалён")
   
    def notify_observers(self):
        """Уведомить всех наблюдателей"""
        print(f"\nПубликация нового сообщения: {self._latest_news}")
        for observer in self._observers:
            observer.update(self._latest_news)
   
    def publish_news(self, news: str):
        """Публикует новости"""
        self._latest_news = news
        self.notify_observers()
   
    def get_observer_count(self) -> int:
        """Возвращает количество наблюдателей"""
        return len(self._observers)

# Конкретный наблюдатель - подписчик электронной почты
class EmailSubscriber(Observer):
    def __init__(self, name: str):
        self.name = name
   
    def update(self, message: str):
        """Получить обновление и отправить электронное письмо"""
        print(f"[Email] {self.name} получил сообщение: {message}")

# Конкретный наблюдатель - SMS-подписчик
class SMSSubscriber(Observer):
    def __init__(self, name: str, phone: str):
        self.name = name
        self.phone = phone
   
    def update(self, message: str):
        """Получить обновление и отправить SMS"""
        print(f"[SMS] {self.phone} - {self.name} получил сообщение: {message}")

# Конкретный наблюдатель - подписчик push-уведомлений приложения
class AppPushSubscriber(Observer):
    def __init__(self, username: str):
        self.username = username
   
    def update(self, message: str):
        """Получить обновление и отправить push-уведомление в приложение"""
        print(f"[Push приложения] Пользователь {self.username} получил push: {message}")

Пример использования

Пример

def demo_basic_observer():
    """Демонстрация базового паттерна Observer"""
    print("=== Демонстрация базового паттерна Observer ==="\n")
   
    # Создать издателя новостей
    publisher = NewsPublisher()
   
    # Создать разные типы подписчиков
    email_user = EmailSubscriber("Чжан Сань")
    sms_user = SMSSubscriber("Ли Сы", "13800138000")
    app_user = AppPushSubscriber("Ван У")
   
    # Зарегистрировать подписчиков
    publisher.register_observer(email_user)
    publisher.register_observer(sms_user)
    publisher.register_observer(app_user)
   
    print(f"\nТекущее количество подписчиков: {publisher.get_observer_count()}")
   
    # Опубликовать новости
    publisher.publish_news("Python 3.12 официально выпущен!")
   
    # Удалить подписчика
    publisher.remove_observer(sms_user)
   
    print(f"\nКоличество подписчиков после удаления: {publisher.get_observer_count()}")
   
    # Снова опубликовать новость
    publisher.publish_news("Руководство по изучению паттерна Observer теперь онлайн!")

# Запустить демонстрацию
if __name__ == "__main__":
    demo_basic_observer()

Вывод:

=== 观察者模式基础演示 ===

观察者 EmailSubscriber 注册成功
观察者 SMSSubscriber 注册成功
观察者 AppPushSubscriber 注册成功

当前订阅者数量: 3

发布新消息: Python 3.12 正式发布！
[邮件] 张三 收到消息: Python 3.12 正式发布！
[短信] 13800138000 - 李四 收到消息: Python 3.12 正式发布！
[应用推送] 用户 王五 收到推送: Python 3.12 正式发布！
观察者 SMSSubscriber 移除成功

移除后订阅者数量: 2

发布新消息: 观察者模式学习指南上线！
[邮件] 张三 收到消息: 观察者模式学习指南上线！
[应用推送] 用户 王五 收到推送: 观察者模式学习指南上线！

Продвинутое применение: система мониторинга цен на акции

Давайте углубим наше понимание на более практичном примере.

Пример

# Пример системы мониторинга акций
class Stock:
    """Класс Stock"""
    def __init__(self, symbol: str, price: float):
        self.symbol = symbol  # Код акции
        self._price = price   # Текущая цена
        self._history = [price]  # История цен
   
    @property
    def price(self) -> float:
        return self._price
   
    @price.setter
    def price(self, value: float):
        """Установить цену и записать историю"""
        old_price = self._price
        self._price = value
        self._history.append(value)
        return old_price, value
   
    def get_price_change(self) -> float:
        """Получить процент изменения цены"""
        if len(self._history) < 2:
            return 0.0
        return ((self._history[-1] - self._history[-2]) / self._history[-2]) * 100

class StockMarket(Subject):
    """Фондовый рынок - конкретный субъект"""
    def __init__(self):
        self._observers: List[Observer] = []
        self._stocks = {}  # Словарь акций: {code: Stock object}
   
    def register_observer(self, observer: Observer):
        if observer not in self_observers:
            self._observers.append(observer)
   
    def remove_observer(self, observer: Observer):
        if observer in self._observers:
            self._observers.remove(observer)
   
    def notify_observers(self, stock: Stock):
        """Уведомить наблюдателей об изменении конкретной акции"""
        change = stock.get_price_change()
        message = f"{stock.symbol}: ${stock.price:.2f} ({change:+.2f}%)"
       
        for observer in self._observers:
            observer.update(message)
   
    def add_stock(self, symbol: str, initial_price: float):
        """Добавить акцию на рынок"""
        self._stocks[symbol] = Stock(symbol, initial_price)
   
    def update_stock_price(self, symbol: str, new_price: float):
        """Обновить цену акции и уведомить наблюдателей"""
        if symbol in self._stocks:
            stock = self._stocks[symbol]
            old_price, current_price = stock.price, new_price
            stock.price = new_price
           
            # Уведомлять только при фактическом изменении цены
            if old_price != current_price:
                self.notify_observers(stock)

class Investor(Observer):
    """Инвестор - Конкретный наблюдатель"""
    def __init__(self, name: str, interest_stocks: List[str] = None):
        self.name = name
        self.interest_stocks = interest_stocks or []
        self.notifications = []
   
    def update(self, message: str):
        """Получать обновления акций"""
        stock_symbol = message.split(":")[0]
       
        # Обращать внимание только на интересующие акции
        if not self.interest_stocks or stock_symbol in self.interest_stocks:
            self.notifications.append(message)
            print(f"Инвестор {self.name} получил: {message}")
   
    def show_notifications(self):
        """Показать все уведомления"""
        print(f"\n=== записи уведомлений {self.name} ===")
        for notification in self.notifications:
            print(f" - {notification}")

def demo_stock_market():
    """Демонстрация фондового рынка"""
    print("\n=== Демонстрация системы мониторинга фондового рынка ===\n")
   
    # Создать фондовый рынок
    market = StockMarket()
   
    # Добавить акции
    market.add_stock("AAPL", 150.0)
    market.add_stock("GOOGL", 2800.0)
    market.add_stock("TSLA", 250.0)
   
    # Создать инвесторов
    investor1 = Investor("Технологический инвестор", ["AAPL", "GOOGL"])
    investor2 = Investor("Инвестор в новую энергетику", ["TSLA"])
    investor3 = Investor("Инвестор всего рынка")
   
    # Регистрация инвесторов
    market.register_observer(investor1)
    market.register_observer(investor2)
    market.register_observer(investor3)
   
    # Симуляция изменений цен акций
    print("Начинаем симуляцию торговли акциями...")
    market.update_stock_price("AAPL", 152.5)   # AAPL растёт
    market.update_stock_price("TSLA", 245.0)   # TSLA падает
    market.update_stock_price("GOOGL", 2820.0) # GOOGL растёт
    market.update_stock_price("AAPL", 151.0)   # AAPL слегка падает
   
    # Отобразить записи уведомлений инвестора
    investor1.show_notifications()
    investor2.show_notifications()
    investor3.show_notifications()

# Запустить демонстрацию фондового рынка
demo_stock_market()

Преимущества и недостатки паттерна «Наблюдатель»

Преимущества

Преимущества Описание
Слабая связанность Зависимость между субъектом и наблюдателями слабая, что позволяет им изменяться независимо
Расширяемость Новых наблюдателей можно легко добавлять, не изменяя код субъекта
Широковещательная связь Поддерживает коммуникацию «один ко многим»; сообщения могут широковещательно рассылаться нескольким объектам
Принцип открытости/закрытости Открыт для расширения, закрыт для изменения

Недостатки

Недостатки Описание Решение
Утечка памяти Если наблюдатели забывают отменять регистрацию, это может привести к утечкам памяти Используйте слабые ссылки или обеспечьте своевременную отмену регистрации
Порядок обновления Порядок обновления наблюдателей не определён Если порядок необходим, можно добавить механизм приоритетов
Проблемы с производительностью Когда наблюдателей много, уведомления могут влиять на производительность Используйте асинхронные уведомления или пакетные обновления

Практические сценарии применения

1. Обработка событий GUI

Пример

# Пример псевдокода
button.on_click(handler1)  # Регистрация наблюдателя события click
button.on_click(handler2)  # Зарегистрировать другого наблюдателя
# При нажатии кнопки будут вызваны все зарегистрированные обработчики

2. Система очередей сообщений

  • Производитель публикует сообщение
  • Несколько потребителей подписываются и обрабатывают сообщения

3. Система мониторинга данных

  • Изменение состояния источника данных
  • Несколько сервисов мониторинга получают уведомления и предпринимают действия

4. Разработка игр

  • Система игровых событий
  • Элементы интерфейса реагируют на изменения состояния игры

Лучшие практики и рекомендации

1. Использование слабых ссылок для предотвращения утечек памяти

Пример

import weakref

class SafeSubject(Subject):
    def __init__(self):
        self._observers = weakref.WeakSet()  # Использовать коллекцию слабых ссылок
   
    def register_observer(self, observer: Observer):
        self._observers.add(observer)

2. Асинхронное уведомление для повышения производительности

Пример

import asyncio

class AsyncObserver(ABC):
    @abstractmethod
    async def update_async(self, message: str):
        pass

class AsyncSubject:
    async def notify_observers_async(self):
        tasks = [observer.update_async(self._message)
                for observer in self._observers]
        await asyncio.gather(*tasks)

3. Добавление обработки ошибок

Пример

def safe_notify(self):
    for observer in self._observers.copy():  # Используйте копию, чтобы избежать проблем с изменением
        try:
            observer.update(self._message)
        except Exception as e:
            print(f"Не удалось обновить наблюдателя {observer}: {e}")

Итог

Паттерн «Наблюдатель» — очень важный паттерн проектирования в Python. Он предоставляет элегантный способ достижения слабой связанности между объектами. После прочтения этой статьи вы должны уметь:

  1. Понимать основные концепции и сценарии применения паттерна «Наблюдатель»
  2. Освоить базовые методы реализации паттерна «Наблюдатель»
  3. Применять паттерн «Наблюдатель» для решения задач в реальных проектах
  4. Избегать распространённых ошибок паттерна «Наблюдатель»
Другие расширения