Python中的观察者模式：从入门到精通

引言

观察者模式允许对象（称为“观察者”）注册到另一个对象（称为“主题”或“被观察者”），从而在主题状态改变时自动收到通知。这种机制使得多个观察者可以独立地响应同一个事件，增强了系统的灵活性和可维护性。特别是在构建高度解耦、易于扩展的应用程序时，观察者模式显得尤为重要。

基础语法介绍

核心概念

• 主题（Subject）：也称作被观察者，负责维护观察者列表，并在自身状态发生变化时通知所有注册的观察者。
• • 观察者（Observer）：是一个对象，其唯一职责是在接收到通知后执行某些操作。

基本语法规则

在Python中实现观察者模式通常涉及到以下几个步骤：

1. 定义一个主题类，该类需要提供添加、删除观察者的方法，以及一个用于通知所有观察者的方法。
2. 2. 创建具体的观察者类，它们实现了一个更新方法，当被主题调用时执行相应操作。
3. 3. 在适当的时候让观察者订阅（即向主题添加自己）或取消订阅（即从主题中移除自己）。

基础实例：天气预报系统

假设我们正在开发一款简单的天气预报应用，其中有一个天气数据源（作为主题）和多个显示设备（作为观察者）。每当天气信息发生变化时，所有连接的设备都应该同步更新自己的显示内容。

class WeatherData:
    def __init__(self):
        self._observers = []
        self._temperature = 0

    def register_observer(self, observer):
        self._observers.append(observer)

    def remove_observer(self, observer):
        self._observers.remove(observer)

    def notify_observers(self):
        for observer in self._observers:
            observer.update(self._temperature)

    def set_temperature(self, temperature):
        self._temperature = temperature
        self.notify_observers()

class DisplayDevice:
    def update(self, temperature):
        print(f"当前温度：{temperature}°C")

# 使用示例
weather_data = WeatherData()
display_device = DisplayDevice()
weather_data.register_observer(display_device)
weather_data.set_temperature(25)  # 输出: 当前温度：25°C

进阶实例：股票交易系统

接下来，让我们看看在更复杂的场景下如何应用观察者模式。例如，在一个股票交易系统中，我们需要实时监控多种股票价格变动，并根据不同用户的策略自动执行买卖操作。

问题描述

用户可以根据个人偏好设置不同的交易策略（如基于移动平均线、相对强弱指数等技术指标），当某只股票的价格达到预设条件时触发相应的交易行为。

高级代码实例

from abc import ABC, abstractmethod

class StockExchange(ABC):
    @abstractmethod
    def register_observer(self, observer):
        pass

    @abstractmethod
    def remove_observer(self, observer):
        pass

    @abstractmethod
    def notify_observers(self):
        pass

class RealStockExchange(StockExchange):
    def __init__(self):
        self._observers = []
        self._stock_prices = {}

    def register_observer(self, observer):
        self._observers.append(observer)

    def remove_observer(self, observer):
        self._observers.remove(observer)

    def notify_observers(self):
        for observer in self._observers:
            observer.update(self._stock_prices)

    def set_price(self, stock_symbol, price):
        self._stock_prices[stock_symbol] = price
        self.notify_observers()

class TradingStrategy(ABC):
    @abstractmethod
    def update(self, stock_prices):
        pass

class MovingAverageStrategy(TradingStrategy):
    def update(self, stock_prices):
        # 实现基于移动平均线的交易逻辑...
        print("根据移动平均线进行交易决策...")

# 使用示例
exchange = RealStockExchange()
strategy = MovingAverageStrategy()
exchange.register_observer(strategy)
exchange.set_price('AAPL', 150.23)  # 触发策略更新

实战案例：社交媒体平台的通知系统

在现实世界的应用中，观察者模式同样发挥着重要作用。比如，在设计社交媒体平台的通知系统时，我们可以利用这一模式来实现用户动态更新与即时推送功能。

问题描述

每当用户发布新内容或与其他用户互动时（如点赞、评论等），平台需要及时将这些信息推送给相关用户，确保他们能够第一时间了解到最新动态。

解决方案与代码实现

我们可以创建一个UserActivity类作为主题，用来存储用户的活动记录；同时定义若干个观察者类（如EmailNotifier、PushNotificationService等），分别处理不同类型的通知方式。

class UserActivity:
    def __init__(self):
        self._observers = []

    def add_observer(self, observer):
        self._observers.append(observer)

    def remove_observer(self, observer):
        self._observers.remove(observer)

    def notify_observers(self, activity):
        for observer in self._observers:
            observer.send_notification(activity)

class EmailNotifier:
    def send_notification(self, activity):
        print(f"发送邮件通知：{activity}")

class PushNotificationService:
    def send_notification(self, activity):
        print(f"发送推送通知：{activity}")

# 使用示例
activity = UserActivity()
email_notifier = EmailNotifier()
push_service = PushNotificationService()
activity.add_observer(email_notifier)
activity.add_observer(push_service)
activity.notify_observers("张三发布了新照片")

扩展讨论

尽管我们已经介绍了观察者模式的基本原理及其在Python中的实现方式，但值得指出的是，该模式也有其局限性和适用范围。例如，在处理大量观察者时可能会导致性能问题；此外，在某些情况下，过度使用观察者模式可能会增加系统的复杂度。因此，在实际应用过程中，开发者需要权衡利弊，合理选择设计模式。