Patrón Observador en Python

El patrón observador es un patrón de diseño de comportamiento que define una relación de dependencia de uno a muchos, permitiendo que múltiples objetos observadores escuchen simultáneamente a un objeto sujeto. Cuando el estado del objeto sujeto cambia, todos sus observadores reciben una notificación y se actualizan automáticamente.

El patrón observador en la vida cotidiana

Imagina el escenario de la suscripción a un periódico:

  • Editorial del periódico(Sujeto) Publica noticias
  • Suscriptores(Observadores) Se suscriben al periódico
  • Cuando se publica un nuevo periódico, todos los suscriptores lo reciben automáticamente
  • Los suscriptores pueden cancelar la suscripción en cualquier momento

Este mecanismo de "publicar-suscribir" es una aplicación típica del patrón observador.

Componentes centrales del patrón observador

El patrón observador incluye cuatro roles clave:

1. Sujeto (Subject)

  • Mantiene una lista de observadores
  • Proporciona métodos para registrar y cancelar el registro de observadores
  • Notifica a todos los observadores sobre cambios de estado

2. Sujeto concreto (Concrete Subject)

  • Almacena información de estado específica
  • Notifica a los observadores cuando cambia el estado

3. Observador (Observer)

  • Define una interfaz de actualización para recibir notificaciones del sujeto

4. Observador concreto (Concrete Observer)

  • Implementa la interfaz de actualización para mantenerse sincronizado con el estado del sujeto

Implementación del patrón observador

Entendamos la implementación del patrón observador mediante un ejemplo completo.

Implementación básica

Ejemplo

from abc import ABC, abstractmethod
from typing import List

# Interfaz del observador
class Observer(ABC):
    @abstractmethod
    def update(self, message: str):
        """Método de actualización del observador"""
        pass

# Interfaz del sujeto
class Subject(ABC):
    @abstractmethod
    def register_observer(self, observer: Observer):
        """Registrar observador"""
        pass
   
    @abstractmethod
    def remove_observer(self, observer: Observer):
        """Eliminar observador"""
        pass
   
    @abstractmethod
    def notify_observers(self):
        """Notificar a todos los observadores"""
        pass

# Sujeto concreto - Publicador de noticias
class NewsPublisher(Subject):
    def __init__(self):
        self._observers: List[Observer] = []
        self._latest_news = ""
   
    def register_observer(self, observer: Observer):
        """Registrar observador"""
        if observer not in self._observers:
            self._observers.append(observer)
            print(f"Observador {observer.__class__.__name__} registrado exitosamente")
   
    def remove_observer(self, observer: Observer):
        """Eliminar observador"""
        if observer in self._observers:
            self._observers.remove(observer)
            print(f"Observador {observer.__class__.__name__} eliminado exitosamente")
   
    def notify_observers(self):
        """Notificar a todos los observadores"""
        print(f"\nPublicar nuevo mensaje: {self._latest_news}")
        for observer in self._observers:
            observer.update(self._latest_news)
   
    def publish_news(self, news: str):
        """Publicar noticias"""
        self._latest_news = news
        self.notify_observers()
   
    def get_observer_count(self) -> int:
        """Obtener número de observadores"""
        return len(self._observers)

# Observador concreto - Suscriptor de correo electrónico
class EmailSubscriber(Observer):
    def __init__(self, name: str):
        self.name = name
   
    def update(self, message: str):
        """Recibir actualización y enviar correo electrónico"""
        print(f"[Correo] {self.name} recibió el mensaje: {message}")

# Observador concreto - Suscriptor de SMS
class SMSSubscriber(Observer):
    def __init__(self, name: str, phone: str):
        self.name = name
        self.phone = phone
   
    def update(self, message: str):
        """Recibir actualización y enviar SMS"""
        print(f"[SMS] {self.phone} - {self.name} recibió el mensaje: {message}")

# Observador concreto - Suscriptor de notificaciones push de la aplicación
class AppPushSubscriber(Observer):
    def __init__(self, username: str):
        self.username = username
   
    def update(self, message: str):
        """Recibir actualización y enviar notificación push de la aplicación"""
        print(f"[Push de la aplicación] El usuario {self.username} recibió la notificación: {message}")

Ejemplo de uso

Ejemplo

def demo_basic_observer():
    """Demostración básica del patrón observador"""
    print("=== Demostración básica del patrón observador ===\n")
   
    # Crear publicador de noticias
    publisher = NewsPublisher()
   
    # Crear diferentes tipos de suscriptores
    email_user = EmailSubscriber("Zhang San")
    sms_user = SMSSubscriber("Li Si", "13800138000")
    app_user = AppPushSubscriber("Wang Wu")
   
    # Registrar suscriptores
    publisher.register_observer(email_user)
    publisher.register_observer(sms_user)
    publisher.register_observer(app_user)
   
    print(f"\nNúmero actual de suscriptores: {publisher.get_observer_count()}")
   
    # Publicar noticias
    publisher.publish_news("¡Python 3.12 publicado oficialmente!")
   
    # Eliminar un suscriptor
    publisher.remove_observer(sms_user)
   
    print(f"\nNúmero de suscriptores después de eliminar: {publisher.get_observer_count()}")
   
    # Publicar noticias nuevamente
    publisher.publish_news("¡Guía de aprendizaje del patrón observador disponible!")

# Ejecutar demostración
if __name__ == "__main__":
    demo_basic_observer()

Resultado de la ejecución:

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

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

当前订阅者数量: 3

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

移除后订阅者数量: 2

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

Aplicación avanzada: Sistema de monitoreo de precios de acciones

Profundicemos nuestra comprensión con un ejemplo más práctico.

Ejemplo

# Ejemplo de sistema de monitoreo de acciones
class Stock:
    """Clase de acción (stock)"""
    def __init__(self, symbol: str, price: float):
        self.symbol = symbol  # Código de la acción
        self._price = price   # Precio actual
        self._history = [price]  # Historial de precios
   
    @property
    def price(self) -> float:
        return self._price
   
    @price.setter
    def price(self, value: float):
        """Establecer precio y registrar historial"""
        old_price = self._price
        self._price = value
        self._history.append(value)
        return old_price, value
   
    def get_price_change(self) -> float:
        """Obtener porcentaje de cambio de precio"""
        if len(self._history) < 2:
            return 0.0
        return ((self._history[-1] - self._history[-2]) / self._history[-2]) * 100

class StockMarket(Subject):
    """Mercado de acciones - Sujeto concreto"""
    def __init__(self):
        self._observers: List[Observer] = []
        self._stocks = {}  # Diccionario de acciones: {código: objeto Stock}
   
    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):
        """Notificar a los observadores cambios en una acción específica"""
        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):
        """Agregar acción al mercado"""
        self._stocks[symbol] = Stock(symbol, initial_price)
   
    def update_stock_price(self, symbol: str, new_price: float):
        """Actualizar el precio de la acción y notificar a los observadores"""
        if symbol in self._stocks:
            stock = self._stocks[symbol]
            old_price, current_price = stock.price, new_price
            stock.price = new_price
           
            # Solo notificar cuando el precio realmente cambia
            if old_price != current_price:
                self.notify_observers(stock)

class Investor(Observer):
    """Inversor - Observador concreto"""
    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):
        """Recibir actualización de acciones"""
        stock_symbol = message.split(":")[0]
       
        # Solo prestar atención a las acciones de interés
        if not self.interest_stocks or stock_symbol in self.interest_stocks:
            self.notifications.append(message)
            print(f"Inversor {self.name} recibió: {message}")
   
    def show_notifications(self):
        """Mostrar todas las notificaciones"""
        print(f"\n=== Registro de notificaciones de {self.name} ===)
        for notification in self.notifications:
            print(f"  - {notification}")

def demo_stock_market():
    """Demostración del mercado de acciones"""
    print("\n=== Demostración del sistema de monitoreo del mercado de acciones ===\n")
   
    # Crear mercado de acciones
    market = StockMarket()
   
    # Agregar acciones
    market.add_stock("AAPL", 150.0)
    market.add_stock("GOOGL", 2800.0)
    market.add_stock("TSLA", 250.0)
   
    # Crear inversores
    investor1 = Investor("Inversor tecnológico", ["AAPL", "GOOGL"])
    investor2 = Investor("Inversor de nuevas energías", ["TSLA"])
    investor3 = Investor("Inversor de todo el mercado")
   
    # Registrar inversores
    market.register_observer(investor1)
    market.register_observer(investor2)
    market.register_observer(investor3)
   
    # Simular cambios en los precios de las acciones
    print("Comenzando la simulación de operaciones bursátiles...")
    market.update_stock_price("AAPL", 152.5)   # AAPL sube
    market.update_stock_price("TSLA", 245.0)   # TSLA baja
    market.update_stock_price("GOOGL", 2820.0) # GOOGL sube
    market.update_stock_price("AAPL", 151.0)   # AAPL baja ligeramente
   
    # Mostrar registros de notificaciones de inversores
    investor1.show_notifications()
    investor2.show_notifications()
    investor3.show_notifications()

# Ejecutar demostración del mercado de acciones
demo_stock_market()

Ventajas y desventajas del patrón observador

Ventajas

Ventaja Descripción
Acoplamiento débil La dependencia entre el sujeto y los observadores es débil, pueden cambiar de forma independiente
Extensibilidad Se pueden agregar fácilmente nuevos observadores sin modificar el código del sujeto
Comunicación por difusión Admite comunicación de uno a muchos; los mensajes se pueden difundir a múltiples objetos
Principio abierto-cerrado Abierto a la extensión, cerrado a la modificación

Desventajas

Desventaja Descripción Solución
Fuga de memoria Si el observador olvida cancelar su registro, puede causar fugas de memoria Usar referencias débiles o asegurarse de cancelar el registro oportunamente
Orden de actualización El orden de actualización de los observadores es indeterminado Si se necesita un orden, se puede agregar un mecanismo de prioridad
Problemas de rendimiento Cuando hay muchos observadores, la notificación puede afectar el rendimiento Usar notificación asíncrona o actualizaciones por lotes

Escenarios de aplicación práctica

1. Manejo de eventos GUI

Ejemplo

# Ejemplo de pseudocódigo
button.on_click(handler1)  # Registrar observador de eventos de clic
button.on_click(handler2)  # Registrar otro observador
# Al hacer clic en el botón, todos los manejadores registrados serán llamados

2. Sistema de cola de mensajes

  • El productor publica mensajes
  • Múltiples consumidores se suscriben y procesan mensajes

3. Sistema de monitoreo de datos

  • Cambio de estado de la fuente de datos
  • Múltiples servicios de monitoreo reciben la notificación y toman medidas

4. Desarrollo de juegos

  • Sistema de eventos del juego
  • Los elementos de la interfaz de usuario (UI) responden a los cambios de estado del juego

Mejores prácticas y consideraciones

1. Usar referencias débiles para evitar fugas de memoria

Ejemplo

import weakref

class SafeSubject(Subject):
    def __init__(self):
        self._observers = weakref.WeakSet()  # Usar un conjunto de referencias débiles
   
    def register_observer(self, observer: Observer):
        self._observers.add(observer)

2. Notificación asíncrona para mejorar el rendimiento

Ejemplo

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. Agregar manejo de errores

Ejemplo

def safe_notify(self):
    for observer in self._observers.copy():  # Usar una copia para evitar problemas de modificación
        try:
            observer.update(self._message)
        except Exception as e:
            print(f"Error al actualizar el observador {observer}: {e}")

Resumen

El patrón observador es un patrón de diseño muy importante en Python, que proporciona una forma elegante de implementar comunicación débilmente acoplada entre objetos. Al estudiar este artículo, deberías poder:

  1. Comprender los conceptos fundamentales del patrón observador y sus escenarios de aplicación
  2. Dominar los métodos básicos de implementación del patrón observador
  3. Aplicar el patrón observador en proyectos reales para resolver problemas
  4. Evitar las trampas comunes del patrón observador
Otras extensiones