Python オブザーバーパターン

オブザーバーパターンは、1対多の依存関係を定義する行動デザインパターンです。複数のオブザーバーオブジェクトが特定のサブジェクトオブジェクトを同時に監視します。サブジェクトオブジェクトの状態が変化すると、そのすべてのオブザーバーが通知を受け取り、自動的に更新されます。

身近なオブザーバーパターン

新聞の購読シーンを想像してみてください：

  • 新聞社（サブジェクト）ニュースを発行する
  • 購読者（オブザーバー）新聞を購読する
  • 新しい新聞が発行されると、すべての購読者は自動的に新聞を受け取ります
  • 購読者はいつでも購読を解除できます

この「発行-購読」の仕組みが、オブザーバーパターンの代表的な応用です。

オブザーバーパターンの核心コンポーネント

オブザーバーパターンには4つの重要な役割があります：

1. サブジェクト（Subject）

  • オブザーバーのリストを管理する
  • オブザーバーの登録・登録解除メソッドを提供する
  • すべてのオブザーバーに状態変化を通知する

2. 具体的サブジェクト（Concrete Subject）

  • 具体的な状態情報を保存する
  • 状態が変化したときにオブザーバーへ通知する

3. オブザーバー（Observer）

  • サブジェクトからの通知を受け取るための更新インターフェースを定義する

4. 具体的オブザーバー（Concrete Observer）

  • 更新インターフェースを実装し、サブジェクトの状態と一致させる

オブザーバーパターンの実装

完全なサンプルを通して、オブザーバーパターンの実装を理解しましょう。

基本実装

サンプル

from abc import ABC, abstractmethod
from typing import List

# オブザーバーインターフェース
class Observer(ABC):
    @abstractmethod
    def update(self, message: str):
        """オブザーバー更新メソッド"""
        pass

# サブジェクトインターフェース
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"[メール] {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}")

# 具体的オブザーバー - アプリ通知購読者
class AppPushSubscriber(Observer):
    def __init__(self, username: str):
        self.username = username
   
    def update(self, message: str):
        """更新を受け取り、アプリ通知を送信"""
        print(f"[アプリ通知] ユーザー {self.username} がプッシュを受信: {message}")

使用例

サンプル

def demo_basic_observer():
    """基本的なオブザーバーパターンのデモ"""
    print("=== オブザーバーパターン基本デモ ===\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 が正式リリースされました！")
   
    # 購読者を1人削除
    publisher.remove_observer(sms_user)
   
    print(f"\n削除後の購読者数: {publisher.get_observer_count()}")
   
    # 再度ニュースを発行
    publisher.publish_news("オブザーバーパターン学習ガイドが公開されました！")

# デモを実行
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:
    """株式クラス"""
    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 = {}  # 株式辞書: {コード: 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):
        """オブザーバーに特定の株式の変化を通知"""
        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対多の通信をサポートし、メッセージを複数のオブジェクトへブロードキャストできます
オープン・クローズドの原則 拡張に対して開かれており、修正に対して閉じられています

短所

短所 説明 解決策
メモリリーク オブザーバーが登録解除を忘れるとメモリリークが発生する可能性があります 弱参照を使用するか、確実に登録解除する
更新順序 オブザーバーの更新順序が不確定 順序が必要な場合は、優先度メカニズムを追加できます
パフォーマンス問題 オブザーバーの数が多いと、通知がパフォーマンスに影響する可能性があります 非同期通知または一括更新を使用する

実際の適用シーン

1. GUIイベント処理

サンプル

# 疑似コード例
button.on_click(handler1)  # クリックイベントのオブザーバーを登録
button.on_click(handler2)  # 別のオブザーバーを登録
# ボタンをクリックすると、登録されたすべてのハンドラーが呼び出される

2. メッセージキューシステム

  • プロデューサーがメッセージを公開する
  • 複数のコンシューマーがメッセージを購読して処理する

3. データ監視システム

  • データソースの状態変化
  • 複数の監視サービスが通知を受け取り、行動を起こす

4. ゲーム開発

  • ゲームイベントシステム
  • UI要素がゲーム状態の変化に応答する

ベストプラクティスと注意事項

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. オブザーバーパターンのよくある落とし穴を回避する
その他の拡張