Что такое паттерн Observer и как его реализовать?

Что такое паттерн Observer и как его реализовать?

Observer (Наблюдатель) — это поведенческий паттерн проектирования, который устанавливает зависимость "один-ко-многим" между объектами. Когда состояние одного объекта (Subject/Observable) изменяется, все его подписчики (Observer'ы) автоматически об этом уведомляются.

Основная идея

Саксофонист (Subject) играет мелодию, а множество слушателей (Observer'ы) слышат изменения в музыке и реагируют на них. Когда саксофонист начинает играть быстрее, каждый слушатель реагирует по-своему.

Компоненты паттерна

1. Subject — объект, состояние которого отслеживается
2. Observer — интерфейс для подписчиков
3. ConcreteObserver — конкретные реализации подписчиков
4. ConcreteSubject — конкретная реализация наблюдаемого объекта

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

from abc import ABC, abstractmethod
from typing import List

# Абстрактный Observer
class Observer(ABC):
    @abstractmethod
    def update(self, subject):
        pass

# Наблюдаемый объект (Subject)
class WeatherStation:
    def __init__(self):
        self._temperature = 0
        self._observers: List[Observer] = []
    
    @property
    def temperature(self):
        return self._temperature
    
    @temperature.setter
    def temperature(self, value):
        self._temperature = value
        self._notify_observers()
    
    def attach(self, observer: Observer):
        """Подписать observer на изменения"""
        if observer not in self._observers:
            self._observers.append(observer)
    
    def detach(self, observer: Observer):
        """Отписать observer"""
        if observer in self._observers:
            self._observers.remove(observer)
    
    def _notify_observers(self):
        """Уведомить всех observers об изменении"""
        for observer in self._observers:
            observer.update(self)

# Конкретные Observer'ы
class TemperatureDisplay(Observer):
    def update(self, subject: WeatherStation):
        print(f"Display: Текущая температура {subject.temperature}°C")

class Alarm(Observer):
    def update(self, subject: WeatherStation):
        if subject.temperature > 30:
            print("Alarm: Опасно высокая температура!")

class Logger(Observer):
    def update(self, subject: WeatherStation):
        print(f"Logger: Записал температуру {subject.temperature}°C в файл")

# Использование
station = WeatherStation()
display = TemperatureDisplay()
alarm = Alarm()
logger = Logger()

station.attach(display)
station.attach(alarm)
station.attach(logger)

station.temperature = 25  # Display и Logger среагируют
station.temperature = 35  # Display, Alarm и Logger среагируют

Современная реализация на Python

В Python для паттерна Observer часто используются обычные функции-колбэки вместо классов:

class EventBus:
    def __init__(self):
        self._subscribers = {}
    
    def subscribe(self, event_type: str, callback):
        """Подписать callback на событие"""
        if event_type not in self._subscribers:
            self._subscribers[event_type] = []
        self._subscribers[event_type].append(callback)
    
    def unsubscribe(self, event_type: str, callback):
        """Отписать callback"""
        if event_type in self._subscribers:
            self._subscribers[event_type].remove(callback)
    
    def emit(self, event_type: str, data=None):
        """Отправить событие всем подписчикам"""
        if event_type in self._subscribers:
            for callback in self._subscribers[event_type]:
                callback(data)

# Использование
bus = EventBus()

def on_user_created(user_data):
    print(f"Отправляем приветственное письмо пользователю {user_data['name']}")

def on_user_created_analytics(user_data):
    print(f"Логируем создание пользователя в аналитику")

bus.subscribe('user:created', on_user_created)
bus.subscribe('user:created', on_user_created_analytics)

# Эмитируем событие
bus.emit('user:created', {'name': 'John', 'email': 'john@example.com'})

Паттерн Observer со слабыми ссылками

Проблема: если забыть отписать observer, он будет жить в памяти. Решение — weakref:

import weakref
from typing import List, Callable

class Subject:
    def __init__(self):
        self._observers: List[weakref.ref] = []
    
    def attach(self, observer):
        """Подписать с использованием слабых ссылок"""
        self._observers.append(weakref.ref(observer, self._remove_observer))
    
    def _remove_observer(self, observer_ref):
        """Автоматически удалить, если observer был удалён"""
        self._observers = [ref for ref in self._observers if ref() is not None]
    
    def notify(self):
        """Уведомить живых observers"""
        for observer_ref in self._observers:
            observer = observer_ref()
            if observer is not None:
                observer.update()

Реальный пример: система событий приложения

from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Callable

class EventType(Enum):
    USER_LOGGED_IN = "user:logged_in"
    USER_LOGGED_OUT = "user:logged_out"
    DATA_UPDATED = "data:updated"

@dataclass
class Event:
    event_type: EventType
    payload: dict

class EventDispatcher:
    def __init__(self):
        self._listeners: Dict[EventType, List[Callable]] = {}
    
    def on(self, event_type: EventType, callback: Callable):
        """Подписать на событие"""
        if event_type not in self._listeners:
            self._listeners[event_type] = []
        self._listeners[event_type].append(callback)
    
    def off(self, event_type: EventType, callback: Callable):
        """Отписать от события"""
        if event_type in self._listeners:
            self._listeners[event_type].remove(callback)
    
    def emit(self, event: Event):
        """Испустить событие"""
        if event.event_type in self._listeners:
            for callback in self._listeners[event.event_type]:
                callback(event)

# Использование
dispatcher = EventDispatcher()

def on_user_login(event: Event):
    print(f"Пользователь {event.payload['username']} залогинился")
    print(f"Отправляем уведомление на {event.payload['email']}")

dispatcher.on(EventType.USER_LOGGED_IN, on_user_login)

event = Event(
    event_type=EventType.USER_LOGGED_IN,
    payload={'username': 'john', 'email': 'john@example.com'}
)
dispatcher.emit(event)

Встроенные инструменты Python

Используй встроенные инструменты, если они подходят:

# Встроенный модуль observer паттерна — это просто функции
def subscribe_to_events(obj, event_name, callback):
    if not hasattr(obj, '_observers'):
        obj._observers = {}
    if event_name not in obj._observers:
        obj._observers[event_name] = []
    obj._observers[event_name].append(callback)

def emit_event(obj, event_name, *args):
    if hasattr(obj, '_observers') and event_name in obj._observers:
        for callback in obj._observers[event_name]:
            callback(*args)

# Для async
import asyncio

class AsyncEventBus:
    def __init__(self):
        self._subscribers = {}
    
    def subscribe(self, event_type: str, callback):
        if event_type not in self._subscribers:
            self._subscribers[event_type] = []
        self._subscribers[event_type].append(callback)
    
    async def emit(self, event_type: str, data=None):
        if event_type in self._subscribers:
            await asyncio.gather(*[
                callback(data) for callback in self._subscribers[event_type]
            ])

Преимущества и недостатки

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

• Слабая связанность между объектами
• Динамическая подписка/отписка
• Легко добавлять новых observers без изменения Subject
• Естественно поддерживает event-driven архитектуру

❌ Недостатки:

• Сложнее отследить в debugger'е
• Может быть утечка памяти (забыл отписать)
• Порядок вызова observers не гарантирован

Когда использовать

• Event-driven системы (GUI, веб-приложения)
• Системы оповещений
• Model-View паттерны (MVC, MVVM)
• Реактивное программирование (RxPy)
• Message brokers (Kafka, RabbitMQ)

Заключение

Observer — это фундаментальный паттерн для:

• Создания слабо связанных систем
• Реализации event-driven архитектур
• Обеспечения динамического взаимодействия между объектами

В Python часто используется в виде простых функций-callback'ов вместо классов, но концепция остаётся та же.