Patron Observateur en Python
Le patron Observateur est un patron de conception comportemental qui définit une relation de dépendance un-à-plusieurs, permettant à plusieurs objets observateurs d'écouter simultanément un certain objet sujet. Lorsque l'état de l'objet sujet change, tous ses observateurs reçoivent une notification et se mettent à jour automatiquement.
Le patron Observateur dans la vie quotidienne
Imaginez un scénario d’abonnement à un journal :
- Bureau du journal(Sujet) publie des actualités
- Les abonnés(Observateur) s’abonnent au journal
- Lorsqu’un nouveau journal est publié, tous les abonnés reçoivent automatiquement le journal
- Les abonnés peuvent annuler leur abonnement à tout moment
Ce mécanisme « publication-abonnement » est une application typique du patron Observateur.
Composants essentiels du patron Observateur
Le patron Observateur contient quatre rôles clés :
1. Sujet
- Maintient une liste d’observateurs
- Fournit des méthodes pour enregistrer et désenregistrer des observateurs
- Notifie tous les observateurs des changements d’état
2. Sujet concret
- Stocke les informations sur l'état concret
- Notifie les observateurs lorsque l'état change
3. Observateur
- Définit une interface de mise à jour pour recevoir les notifications du sujet
4. Observateur concret
- Implémente l'interface de mise à jour pour rester cohérent avec l'état du sujet
Implémentation du patron Observateur
Comprenons l'implémentation du patron Observateur à travers un exemple complet.
Implémentation de base
Exemple
from abc import ABC, abstractmethod
from typing import List
# Interface Observateur
class Observer(ABC):
@abstractmethod
def update(self, message: str):
"""Méthode de mise à jour de l'observateur"""
pass
# Interface Sujet
class Subject(ABC):
@abstractmethod
def register_observer(self, observer: Observer):
"""Enregistrer l'observateur"""
pass
@abstractmethod
def remove_observer(self, observer: Observer):
"""Supprimer l'observateur"""
pass
@abstractmethod
def notify_observers(self):
"""Notifier tous les observateurs"""
pass
# Sujet concret - Éditeur de nouvelles
class NewsPublisher(Subject):
def __init__(self):
self._observers: List[Observer] = []
self._latest_news = ""
def register_observer(self, observer: Observer):
"""Enregistrer l'observateur"""
if observer not in self._observers:
self._observers.append(observer)
print(f"Observateur {observer.__class__.__name__} enregistré avec succès")
def remove_observer(self, observer: Observer):
"""Supprimer l'observateur"""
if observer in self._observers:
self._observers.remove(observer)
print(f"Observateur {observer.__class__.__name__} supprimé avec succès")
def notify_observers(self):
"""Notifier tous les observateurs"""
print(f"\nPublier un nouveau message : {self._latest_news}")
for observer in self._observers:
observer.update(self._latest_news)
def publish_news(self, news: str):
"""Publier les actualités"""
self._latest_news = news
self.notify_observers()
def get_observer_count(self) -> int:
"""Obtenir le nombre d'observateurs"""
return len(self._observers)
# Observateur concret - Abonné e-mail
class EmailSubscriber(Observer):
def __init__(self, name: str):
self.name = name
def update(self, message: str):
"""Recevoir une mise à jour et envoyer un e-mail"""
print(f"[E-mail] {self.name} a reçu le message : {message}")
# Observateur concret - Abonné SMS
class SMSSubscriber(Observer):
def __init__(self, name: str, phone: str):
self.name = name
self.phone = phone
def update(self, message: str):
"""Recevoir une mise à jour et envoyer un SMS"""
print(f"[SMS] {self.phone} - {self.name} a reçu le message : {message}")
# Observateur concret - Abonné aux notifications push de l'application
class AppPushSubscriber(Observer):
def __init__(self, username: str):
self.username = username
def update(self, message: str):
"""Recevoir la mise à jour et envoyer une notification push d'application"""
print(f"[Push d'application] L'utilisateur {self.username} a reçu une notification push : {message}")
from typing import List
# Interface Observateur
class Observer(ABC):
@abstractmethod
def update(self, message: str):
"""Méthode de mise à jour de l'observateur"""
pass
# Interface Sujet
class Subject(ABC):
@abstractmethod
def register_observer(self, observer: Observer):
"""Enregistrer l'observateur"""
pass
@abstractmethod
def remove_observer(self, observer: Observer):
"""Supprimer l'observateur"""
pass
@abstractmethod
def notify_observers(self):
"""Notifier tous les observateurs"""
pass
# Sujet concret - Éditeur de nouvelles
class NewsPublisher(Subject):
def __init__(self):
self._observers: List[Observer] = []
self._latest_news = ""
def register_observer(self, observer: Observer):
"""Enregistrer l'observateur"""
if observer not in self._observers:
self._observers.append(observer)
print(f"Observateur {observer.__class__.__name__} enregistré avec succès")
def remove_observer(self, observer: Observer):
"""Supprimer l'observateur"""
if observer in self._observers:
self._observers.remove(observer)
print(f"Observateur {observer.__class__.__name__} supprimé avec succès")
def notify_observers(self):
"""Notifier tous les observateurs"""
print(f"\nPublier un nouveau message : {self._latest_news}")
for observer in self._observers:
observer.update(self._latest_news)
def publish_news(self, news: str):
"""Publier les actualités"""
self._latest_news = news
self.notify_observers()
def get_observer_count(self) -> int:
"""Obtenir le nombre d'observateurs"""
return len(self._observers)
# Observateur concret - Abonné e-mail
class EmailSubscriber(Observer):
def __init__(self, name: str):
self.name = name
def update(self, message: str):
"""Recevoir une mise à jour et envoyer un e-mail"""
print(f"[E-mail] {self.name} a reçu le message : {message}")
# Observateur concret - Abonné SMS
class SMSSubscriber(Observer):
def __init__(self, name: str, phone: str):
self.name = name
self.phone = phone
def update(self, message: str):
"""Recevoir une mise à jour et envoyer un SMS"""
print(f"[SMS] {self.phone} - {self.name} a reçu le message : {message}")
# Observateur concret - Abonné aux notifications push de l'application
class AppPushSubscriber(Observer):
def __init__(self, username: str):
self.username = username
def update(self, message: str):
"""Recevoir la mise à jour et envoyer une notification push d'application"""
print(f"[Push d'application] L'utilisateur {self.username} a reçu une notification push : {message}")
Exemple d'utilisation
Exemple
def demo_basic_observer():
"""Démonstration de base du patron Observateur"""
print("=== Démonstration de base du patron Observateur ==="\n")
# Créer un éditeur de nouvelles
publisher = NewsPublisher()
# Créer différents types d'abonnés
email_user = EmailSubscriber("Zhang San")
sms_user = SMSSubscriber("Li Si", "13800138000")
app_user = AppPushSubscriber("Wang Wu")
# Enregistrer les abonnés
publisher.register_observer(email_user)
publisher.register_observer(sms_user)
publisher.register_observer(app_user)
print(f"\nNombre actuel d'abonnés : {publisher.get_observer_count()}")
# Publier les actualités
publisher.publish_news("Python 3.12 officiellement publié !")
# Supprimer un abonné
publisher.remove_observer(sms_user)
print(f"\nNombre d'abonnés après suppression : {publisher.get_observer_count()}")
# Publier à nouveau des actualités
publisher.publish_news("Le guide d'apprentissage du patron Observer est en ligne !")
# Exécuter la démonstration
if __name__ == "__main__":
demo_basic_observer()
"""Démonstration de base du patron Observateur"""
print("=== Démonstration de base du patron Observateur ==="\n")
# Créer un éditeur de nouvelles
publisher = NewsPublisher()
# Créer différents types d'abonnés
email_user = EmailSubscriber("Zhang San")
sms_user = SMSSubscriber("Li Si", "13800138000")
app_user = AppPushSubscriber("Wang Wu")
# Enregistrer les abonnés
publisher.register_observer(email_user)
publisher.register_observer(sms_user)
publisher.register_observer(app_user)
print(f"\nNombre actuel d'abonnés : {publisher.get_observer_count()}")
# Publier les actualités
publisher.publish_news("Python 3.12 officiellement publié !")
# Supprimer un abonné
publisher.remove_observer(sms_user)
print(f"\nNombre d'abonnés après suppression : {publisher.get_observer_count()}")
# Publier à nouveau des actualités
publisher.publish_news("Le guide d'apprentissage du patron Observer est en ligne !")
# Exécuter la démonstration
if __name__ == "__main__":
demo_basic_observer()
Sortie :
=== 观察者模式基础演示 === 观察者 EmailSubscriber 注册成功 观察者 SMSSubscriber 注册成功 观察者 AppPushSubscriber 注册成功 当前订阅者数量: 3 发布新消息: Python 3.12 正式发布！ [邮件] 张三 收到消息: Python 3.12 正式发布！ [短信] 13800138000 - 李四 收到消息: Python 3.12 正式发布！ [应用推送] 用户 王五 收到推送: Python 3.12 正式发布！ 观察者 SMSSubscriber 移除成功 移除后订阅者数量: 2 发布新消息: 观察者模式学习指南上线！ [邮件] 张三 收到消息: 观察者模式学习指南上线！ [应用推送] 用户 王五 收到推送: 观察者模式学习指南上线！
Application avancée : Système de surveillance des cours boursiers
Approfondissons notre compréhension grâce à un exemple plus pratique.
Exemple
# Exemple de système de surveillance des actions
class Stock:
"""Classe Stock"""
def __init__(self, symbol: str, price: float):
self.symbol = symbol # Code de l'action
self._price = price # Prix actuel
self._history = [price] # Historique des prix
@property
def price(self) -> float:
return self._price
@price.setter
def price(self, value: float):
"""Définir le prix et enregistrer l'historique"""
old_price = self._price
self._price = value
self._history.append(value)
return old_price, value
def get_price_change(self) -> float:
"""Obtenir le pourcentage de variation du prix"""
if len(self._history) < 2:
return 0.0
return ((self._history[-1] - self._history[-2]) / self._history[-2]) * 100
class StockMarket(Subject):
"""Marché boursier - Sujet concret"""
def __init__(self):
self._observers: List[Observer] = []
self._stocks = {} # Dictionnaire des actions : {code: objet 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):
"""Notifier les observateurs de la variation d'une action spécifique"""
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):
"""Ajouter une action au marché"""
self._stocks[symbol] = Stock(symbol, initial_price)
def update_stock_price(self, symbol: str, new_price: float):
"""Mettre à jour le prix d'une action et notifier les observateurs"""
if symbol in self._stocks:
stock = self._stocks[symbol]
old_price, current_price = stock.price, new_price
stock.price = new_price
# Notifier uniquement lorsque le prix change réellement
if old_price != current_price:
self.notify_observers(stock)
class Investor(Observer):
"""Investisseur - Observateur concret"""
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):
"""Recevoir les mises à jour des actions"""
stock_symbol = message.split(":")[0]
# Ne faire attention qu'aux actions d'intérêt
if not self.interest_stocks or stock_symbol in self.interest_stocks:
self.notifications.append(message)
print(f"L'investisseur {self.name} a reçu : {message}")
def show_notifications(self):
"""Afficher toutes les notifications"""
print(f"\n=== Historique des notifications de {self.name} ===")
for notification in self.notifications:
print(f" - {notification}")
def demo_stock_market():
"""Démonstration du marché boursier"""
print("\n=== Démonstration du système de surveillance du marché boursier ===\n")
# Créer un marché boursier
market = StockMarket()
# Ajouter des actions
market.add_stock("AAPL", 150.0)
market.add_stock("GOOGL", 2800.0)
market.add_stock("TSLA", 250.0)
# Créer des investisseurs
investor1 = Investor("Investisseur en technologie", ["AAPL", "GOOGL"])
investor2 = Investor("Investisseur en énergies nouvelles", ["TSLA"])
investor3 = Investor("Investisseur sur l'ensemble du marché")
# Enregistrer les investisseurs
market.register_observer(investor1)
market.register_observer(investor2)
market.register_observer(investor3)
# Simuler les variations du cours des actions
print("Commencer la simulation du trading d'actions...")
market.update_stock_price("AAPL", 152.5) # AAPL augmente
market.update_stock_price("TSLA", 245.0) # TSLA baisse
market.update_stock_price("GOOGL", 2820.0) # GOOGL augmente
market.update_stock_price("AAPL", 151.0) # AAPL baisse légèrement
# Afficher les enregistrements de notification des investisseurs
investor1.show_notifications()
investor2.show_notifications()
investor3.show_notifications()
# Exécuter la démonstration du marché boursier
demo_stock_market()
class Stock:
"""Classe Stock"""
def __init__(self, symbol: str, price: float):
self.symbol = symbol # Code de l'action
self._price = price # Prix actuel
self._history = [price] # Historique des prix
@property
def price(self) -> float:
return self._price
@price.setter
def price(self, value: float):
"""Définir le prix et enregistrer l'historique"""
old_price = self._price
self._price = value
self._history.append(value)
return old_price, value
def get_price_change(self) -> float:
"""Obtenir le pourcentage de variation du prix"""
if len(self._history) < 2:
return 0.0
return ((self._history[-1] - self._history[-2]) / self._history[-2]) * 100
class StockMarket(Subject):
"""Marché boursier - Sujet concret"""
def __init__(self):
self._observers: List[Observer] = []
self._stocks = {} # Dictionnaire des actions : {code: objet 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):
"""Notifier les observateurs de la variation d'une action spécifique"""
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):
"""Ajouter une action au marché"""
self._stocks[symbol] = Stock(symbol, initial_price)
def update_stock_price(self, symbol: str, new_price: float):
"""Mettre à jour le prix d'une action et notifier les observateurs"""
if symbol in self._stocks:
stock = self._stocks[symbol]
old_price, current_price = stock.price, new_price
stock.price = new_price
# Notifier uniquement lorsque le prix change réellement
if old_price != current_price:
self.notify_observers(stock)
class Investor(Observer):
"""Investisseur - Observateur concret"""
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):
"""Recevoir les mises à jour des actions"""
stock_symbol = message.split(":")[0]
# Ne faire attention qu'aux actions d'intérêt
if not self.interest_stocks or stock_symbol in self.interest_stocks:
self.notifications.append(message)
print(f"L'investisseur {self.name} a reçu : {message}")
def show_notifications(self):
"""Afficher toutes les notifications"""
print(f"\n=== Historique des notifications de {self.name} ===")
for notification in self.notifications:
print(f" - {notification}")
def demo_stock_market():
"""Démonstration du marché boursier"""
print("\n=== Démonstration du système de surveillance du marché boursier ===\n")
# Créer un marché boursier
market = StockMarket()
# Ajouter des actions
market.add_stock("AAPL", 150.0)
market.add_stock("GOOGL", 2800.0)
market.add_stock("TSLA", 250.0)
# Créer des investisseurs
investor1 = Investor("Investisseur en technologie", ["AAPL", "GOOGL"])
investor2 = Investor("Investisseur en énergies nouvelles", ["TSLA"])
investor3 = Investor("Investisseur sur l'ensemble du marché")
# Enregistrer les investisseurs
market.register_observer(investor1)
market.register_observer(investor2)
market.register_observer(investor3)
# Simuler les variations du cours des actions
print("Commencer la simulation du trading d'actions...")
market.update_stock_price("AAPL", 152.5) # AAPL augmente
market.update_stock_price("TSLA", 245.0) # TSLA baisse
market.update_stock_price("GOOGL", 2820.0) # GOOGL augmente
market.update_stock_price("AAPL", 151.0) # AAPL baisse légèrement
# Afficher les enregistrements de notification des investisseurs
investor1.show_notifications()
investor2.show_notifications()
investor3.show_notifications()
# Exécuter la démonstration du marché boursier
demo_stock_market()
Avantages et inconvénients du patron Observateur
Avantages
| Avantages | Description |
|---|---|
| Couplage faible | La dépendance entre le sujet et les observateurs est faible, ce qui leur permet de changer indépendamment |
| Extensibilité | De nouveaux observateurs peuvent être ajoutés facilement sans modifier le code du sujet |
| Communication par diffusion | Prend en charge une communication un-à-plusieurs ; les messages peuvent être diffusés vers plusieurs objets |
| Principe ouvert/fermé | Ouvert à l'extension, fermé à la modification |
Inconvénients
| Inconvénients | Description | Solution |
|---|---|---|
| Fuite de mémoire | Les observateurs qui oublient de se désinscrire peuvent provoquer des fuites de mémoire | Utilisez des références faibles ou assurez-vous d'une désinscription en temps voulu |
| Ordre de mise à jour | L'ordre de mise à jour des observateurs est incertain | Si un ordre est nécessaire, un mécanisme de priorité peut être ajouté |
| Problèmes de performance | Lorsqu'il y a de nombreux observateurs, les notifications peuvent affecter les performances | Utilisez des notifications asynchrones ou des mises à jour par lots |
Scénarios d'application pratiques
1. Gestion des événements GUI
Exemple
# Exemple de pseudo-code
button.on_click(handler1) # Enregistrer l'observateur d'événement de clic
button.on_click(handler2) # Enregistrer un autre observateur
# Lorsque le bouton est cliqué, tous les gestionnaires enregistrés seront appelés
button.on_click(handler1) # Enregistrer l'observateur d'événement de clic
button.on_click(handler2) # Enregistrer un autre observateur
# Lorsque le bouton est cliqué, tous les gestionnaires enregistrés seront appelés
2. Système de file d'attente de messages
- Le producteur publie un message
- Plusieurs consommateurs s'abonnent et traitent les messages
3. Système de surveillance des données
- Changements d'état de la source de données
- Plusieurs services de surveillance reçoivent des notifications et prennent des mesures
4. Développement de jeux
- Système d'événements de jeu
- Les éléments d'interface utilisateur réagissent aux changements d'état du jeu
Bonnes pratiques et considérations
1. Utiliser des références faibles pour éviter les fuites de mémoire
Exemple
import weakref
class SafeSubject(Subject):
def __init__(self):
self._observers = weakref.WeakSet() # Utiliser une collection de références faibles
def register_observer(self, observer: Observer):
self._observers.add(observer)
class SafeSubject(Subject):
def __init__(self):
self._observers = weakref.WeakSet() # Utiliser une collection de références faibles
def register_observer(self, observer: Observer):
self._observers.add(observer)
2. Notification asynchrone pour améliorer les performances
Exemple
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)
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. Ajouter la gestion des erreurs
Exemple
def safe_notify(self):
for observer in self._observers.copy(): # Utiliser une copie pour éviter les problèmes de modification
try:
observer.update(self._message)
except Exception as e:
print(f"La mise à jour de l'observateur {observer} a échoué : {e}")
for observer in self._observers.copy(): # Utiliser une copie pour éviter les problèmes de modification
try:
observer.update(self._message)
except Exception as e:
print(f"La mise à jour de l'observateur {observer} a échoué : {e}")
Résumé
Le patron Observateur est un modèle de conception très important en Python. Il offre une manière élégante de parvenir à un couplage lâche entre les objets. Après avoir lu cet article, vous devriez être capable de :
- Comprendre les concepts de base et les scénarios d'application du patron Observateur
- Maîtriser les méthodes d'implémentation de base du patron Observateur
- Appliquer le patron Observateur pour résoudre des problèmes dans des projets réels
- Éviter les pièges courants du patron Observateur