观察者模式是设计模式之一，实现一对多依赖，当主题状态变化时通知所有观察者更新。在Python中，通过自定义接口或内置模块实现观察者模式，可提高程序灵活性和扩展性，尤其适用于状态变化时触发操作的场景，如事件驱动系统、数据同步、发布/订阅系统和气象监测系统等。观察者模式具有松耦合、灵活、可复用的优点，但也需注意其可能带来的开闭原则挑战、效率、循环依赖等问题。
一、引言
观察者模式（Observer Pattern）是设计模式中的一种行为型模式，它定义了一种一对多的依赖关系，让多个观察者对象同时监听某一主题对象，当主题对象状态发生改变时，会通知所有观察者对象，使它们能够自动更新自己。这种模式使得目标和观察者之间松散耦合，因为观察者只知道他们需要关注的主题发生了变化，而不必了解变化的具体细节。
在Python编程中，观察者模式可以极大地提高程序的可扩展性和灵活性，尤其适用于那些需要在状态变化时执行特定操作的场景。
二、观察者模式的实现
在Python中，我们可以使用内置的collections.abc模块中的MutableSequence抽象基类，或者自定义一个简单的接口来实现观察者模式。下面是一个基于自定义接口的基本实现：
import abc# 定义观察者接口
class Observer(metaclass=abc.ABCMeta):@abc.abstractmethoddef update(self, subject, message):pass# 定义被观察者（主题）类
class Subject:def __init__(self):self._observers = []def attach(self, observer):self._observers.append(observer)def detach(self, observer):self._observers.remove(observer)def notify(self, message):for observer in self._observers:observer.update(self, message)# 实现具体的观察者
class ConcreteObserver(Observer):def update(self, subject, message):print(f"接收到主题({subject})的通知: {message}")# 使用示例
if __name__ == "__main__":subject = Subject()observer1 = ConcreteObserver()observer2 = ConcreteObserver()subject.attach(observer1)subject.attach(observer2)subject.notify("状态已更新")三、观察者模式的应用场景
3.1 事件驱动系统
观察者模式可以很好地应用于事件驱动编程，尤其是在构建GUI应用程序、网络I/O处理和异步任务调度等领域。下面是一个简化的事件驱动编程中的观察者模式实现，模拟了一个简单的事件总线（Event Bus）系统：
import abc
from typing import Callableclass Event(metaclass=abc.ABCMeta):passclass MyCustomEvent(Event):def __init__(self, data):self.data = dataclass EventHandler(metaclass=abc.ABCMeta):@abc.abstractmethoddef handle(self, event: Event):passclass EventBus:def __init__(self):self._handlers = {}def register_handler(self, event_type: type, handler: EventHandler):if event_type not in self._handlers:self._handlers[event_type] = []self._handlers[event_type].append(handler)def unregister_handler(self, event_type: type, handler: EventHandler):if event_type in self._handlers and handler in self._handlers[event_type]:self._handlers[event_type].remove(handler)def publish(self, event: Event):event_type = type(event)if event_type in self._handlers:for handler in self._handlers[event_type]:handler.handle(event)# 具体的事件处理器实现
class CustomEventHandler(EventHandler):def handle(self, event: Event):if isinstance(event, MyCustomEvent):print(f"处理自定义事件，数据为: {event.data}")# 应用场景
if __name__ == "__main__":bus = EventBus()custom_handler = CustomEventHandler()# 注册事件处理器bus.register_handler(MyCustomEvent, custom_handler)# 发布事件custom_event = MyCustomEvent("Hello, World!")bus.publish(custom_event)在这个例子中，EventBus作为一个中心组件，负责管理和分发事件。EventHandler是观察者接口，任何想要处理事件的类都需要实现这个接口。当有事件发布到事件总线上时，事件总线会查找对应类型的事件处理器，并调用它们的handle方法来处理事件。
MyCustomEvent是具体的事件类型，而CustomEventHandler则是针对该事件类型的具体处理器。在主程序中，我们创建了一个事件总线实例，注册了一个自定义事件处理器，然后发布了一个自定义事件。发布事件后，自定义事件处理器接收到事件并进行了处理。这就是观察者模式在事件驱动编程中的基本应用。
3.2 数据同步
在数据同步的场景中，观察者模式可以用来实现在数据源发生更改时，自动更新所有依赖于该数据的实体。以下是一个简单的Python实现，模拟了一个数据库表数据变化时通知多个订阅者进行数据同步的例子：
import abc# 定义观察者接口
class DataSyncObserver(metaclass=abc.ABCMeta):@abc.abstractmethoddef sync_data(self, updated_data):pass# 定义被观察者（数据源）
class DataSource:def __init__(self):self._observers = []def attach(self, observer: DataSyncObserver):self._observers.append(observer)def detach(self, observer: DataSyncObserver):self._observers.remove(observer)def update_data(self, new_data):self._data = new_dataself.notify_observers()def notify_observers(self):for observer in self._observers:observer.sync_data(self._data)# 实现具体的数据同步观察者
class CacheUpdater(DataSyncObserver):def __init__(self, cache):self.cache = cachedef sync_data(self, updated_data):print(f"CacheUpdater: Updating cache with new data: {updated_data}")self.cache.update(updated_data)class FrontendUpdater(DataSyncObserver):def __init__(self, frontend):self.frontend = frontenddef sync_data(self, updated_data):print(f"FrontendUpdater: Updating frontend with new data: {updated_data}")self.frontend.render(updated_data)# 应用场景
if __name__ == "__main__":source = DataSource()cache_updater = CacheUpdater(Cache())frontend_updater = FrontendUpdater(Frontend())source.attach(cache_updater)source.attach(frontend_updater)# 数据源数据发生更改source.update_data({"key": "value"})# 这里假设有Cache和Frontend类，分别代表缓存和前端展示组件
# 并且它们都有update方法用于更新数据
在这个例子中，DataSource是被观察者，它维护了一个观察者列表，并在数据更新时通过notify_observers方法通知所有观察者。CacheUpdater和FrontendUpdater都是数据同步观察者，它们实现了sync_data方法，当接收到数据源的更新通知时，会同步更新自身的缓存或前端显示数据。当数据源的update_data方法被调用时，所有已注册的观察者都会收到更新通知并执行同步操作。
3.3 发布/订阅系统
在发布/订阅（Publish/Subscribe）系统中，观察者模式是非常适合的，因为它可以很好地处理一对多的关系，即一个主题（发布者）可以有多个订阅者（观察者）。发布者发布消息时，所有订阅该主题的观察者都会收到通知。以下是一个基于Python实现的简化的发布/订阅系统示例：
import abc# 定义消息接口
class Message(metaclass=abc.ABCMeta):@abc.abstractmethoddef get_message(self) -> str:pass# 定义订阅者接口
class Subscriber(metaclass=abc.ABCMeta):@abc.abstractmethoddef receive_message(self, message: Message):pass# 定义发布/订阅系统的核心类
class PubSubSystem:def __init__(self):self._subscribers = {}def subscribe(self, topic: str, subscriber: Subscriber):if topic not in self._subscribers:self._subscribers[topic] = []self._subscribers[topic].append(subscriber)def unsubscribe(self, topic: str, subscriber: Subscriber):if topic in self._subscribers and subscriber in self._subscribers[topic]:self._subscribers[topic].remove(subscriber)def publish(self, message: Message):topic = message.get_topic()if topic in self._subscribers:for subscriber in self._subscribers[topic]:subscriber.receive_message(message)# 实现具体的消息类和订阅者类
class NewsMessage(Message):def __init__(self, title, content):self.title = titleself.content = contentdef get_topic(self) -> str:return "news"def get_message(self) -> str:return f"Title: {self.title}, Content: {self.content}"class NewsSubscriber(Subscriber):def receive_message(self, message: Message):if isinstance(message, NewsMessage):print(f"NewsSubscriber received news: {message.get_message()}")# 应用场景
if __name__ == "__main__":pubsub = PubSubSystem()subscriber = NewsSubscriber()pubsub.subscribe("news", subscriber)news_message = NewsMessage("Breaking News!", "Important event occurred.")pubsub.publish(news_message)
在这个示例中，PubSubSystem扮演了发布者的角色，它维护了一个字典，键是主题名，值是订阅该主题的观察者列表。当有新的消息发布时，系统会查找订阅了该主题的所有订阅者，并调用它们的receive_message方法。NewsMessage是一个具体的消息类，NewsSubscriber是订阅者类，它定义了如何处理特定类型的消息。在主程序中，我们创建了一个发布/订阅系统实例，订阅了“news”主题，并发布了一个新闻消息。发布后，订阅了该主题的订阅者将会接收到并处理这条消息。
3.4 气象监测系统
在气象监测系统中，我们可以利用观察者模式来实现气象数据变化时自动通知订阅者（如气象分析软件、预警系统等）。下面是一个简化的Python实现：
import abc# 定义气象数据接口
class WeatherData(metaclass=abc.ABCMeta):@abc.abstractmethoddef get_temperature(self) -> float:pass@abc.abstractmethoddef get_humidity(self) -> float:pass@abc.abstractmethoddef get_pressure(self) -> float:pass# 定义观察者接口
class WeatherObserver(metaclass=abc.ABCMeta):@abc.abstractmethoddef update(self, weather_data: WeatherData):pass# 定义气象监测站（被观察者）
class WeatherStation:def __init__(self):self._observers = []self._weather_data = WeatherDataImpl() # 假定WeatherDataImpl是实现了WeatherData接口的具体气象数据类def register_observer(self, observer: WeatherObserver):self._observers.append(observer)def remove_observer(self, observer: WeatherObserver):self._observers.remove(observer)def measurements_changed(self):for observer in self._observers:observer.update(self._weather_data)# 实现具体的观察者类
class WeatherAlertSystem(WeatherObserver):def update(self, weather_data: WeatherData):temp = weather_data.get_temperature()humidity = weather_data.get_humidity()pressure = weather_data.get_pressure()# 模拟检查天气状况并采取行动if temp > 35 or pressure < 950:print("Weather Alert System: Sending alerts due to extreme conditions!")# 应用场景
if __name__ == "__main__":station = WeatherStation()alert_system = WeatherAlertSystem()# 注册观察者station.register_observer(alert_system)# 模拟气象数据变化# （通常这部分由传感器或其他实时数据源提供，这里仅做演示）fake_update = WeatherDataImpl(temp=38, humidity=80, pressure=940)station._weather_data = fake_update # 更改气象数据# 通知所有观察者数据已改变station.measurements_changed()
在这个例子中，WeatherStation是被观察者，它维护一个观察者列表，并在气象数据变化时调用measurements_changed方法通知所有观察者。WeatherObserver是观察者接口，具体观察者（如WeatherAlertSystem）实现了该接口以处理气象数据变化的事件。当气象监测站接收到新的气象数据时，所有已注册的观察者都会收到通知并执行相应的更新操作。
四、优缺点及适用情况
4.1 优缺点
- 优点：
- 松耦合：观察者模式降低了主题（被观察者）和观察者之间的耦合度，它们只需要知道对方的接口即可，无需了解对方的具体实现细节。
- 灵活性：观察者可以在运行时动态添加或删除，增强了系统的灵活性和可扩展性。
- 复用性：一个被观察者可以同时拥有任意数量的观察者，方便了系统的模块化设计和重用。
- 广播通知：当被观察者状态改变时，能够一次性通知所有相关的观察者，简化了状态传播的过程。
- 缺点：
- 开闭原则的挑战：为了支持新的观察者，有时需要修改被观察者的代码以添加新的通知方法。
- 效率问题：当观察者数量庞大且频繁更新时，通知所有观察者的操作可能成为性能瓶颈。
- 循环依赖：如果没有正确设计，可能会导致观察者和被观察者之间的循环依赖问题。
- 控制复杂度：随着系统规模增大，管理观察者列表和通知过程可能变得复杂，特别是在有多种类型的观察者和通知策略的情况下。
- 通知顺序无法保证：观察者模式并不保证通知的顺序，如果需要有序的通知，可能需要结合其他设计模式或机制来实现。
此外，Python标准库也提供了threading模块的Event、Condition和Queue等机制，用于线程间通信和事件通知，可以根据具体应用场景考虑是否采用这些内建工具替代传统的观察者模式。
4.2 实战中的注意事项
在实际应用观察者模式时，有几个关键点需要注意：
-
避免循环依赖：确保不会出现观察者依赖于被观察者，又被被观察者依赖的情况，否则可能导致无限循环通知。
-
处理好资源管理：尤其是当观察者数量巨大或者生命周期不确定时，要妥善处理资源释放，防止内存泄露等问题。
-
控制通知频率：如果观察者数量众多并且频繁更新，可能会影响性能。可通过批处理、节流或截流等手段控制通知的频率和时机。
-
明确边界和责任：清晰定义被观察者和观察者的职责范围，避免二者职责混淆，造成设计上的混乱。
五、高级特性和优化
在实际项目中，观察者模式可以通过一些优化手段来增强其表现力和实用性。例如：
5.1 异步通知
在高并发或多线程环境中，被观察者发出通知时，可能需要使用异步方式来避免阻塞。Python的asyncio库可以帮助我们将通知过程转换为非阻塞式任务，观察者则通过异步回调函数处理更新。
import asyncioclass AsyncStockPrice(Subject):async def notify_observers_async(self, price):tasks = [observer.update(self, price) for observer in self._observers]await asyncio.gather(*tasks)class AsyncPriceSubscriber(Observer):async def update(self, subject, price):print(f"异步股票价格更新：{subject.symbol}当前价格为{price:.2f}")# 异步示例
loop = asyncio.get_event_loop()
async_stock = AsyncStockPrice('AAPL')
async_subscriber1 = AsyncPriceSubscriber()
async_subscriber2 = AsyncPriceSubscriber()async_stock.attach(async_subscriber1)
async_stock.attach(async_subscriber2)loop.run_until_complete(async_stock.notify_observers_async(147.45))
5.2 弱引用
在长期运行的应用中，为了避免内存泄漏，可以使用弱引用（weakref）来持有观察者引用。这样，当没有其他强引用指向观察者时，系统会在适当的时候自动回收资源。
import weakrefclass WeakReferenceObserverPattern:def __init__(self):self._observers = weakref.WeakSet()def attach(self, observer):self._observers.add(observer)def detach(self, observer):try:self._observers.remove(observer)except KeyError:passdef notify_observers(self, *args, **kwargs):observers_copy = list(self._observers)for observer in observers_copy:observer.update(self, *args, **kwargs)
5.3 过滤器和中介者
在大型系统中，为了更精细地控制事件传递和处理流程，观察者模式可能结合过滤器模式或者中介者模式使用，以控制哪些观察者应该接收到什么样的更新。过滤器可以根据预设条件筛选出需要接收更新的观察者，而中介者则可以集中处理和转发事件，从而简化对象之间的交互。下面是一个简单示例，展示了如何在观察者模式中加入过滤器和中介者概念：
from abc import ABC, abstractmethod
from collections import defaultdictclass Filter(metaclass=ABC):@abstractmethoddef should_notify(self, observer, subject, change):passclass Mediator:def __init__(self):self._observers = defaultdict(list)self._filters = []def register_observer(self, observer, filter_=None):if filter_ is not None:self._filters.append(filter_)self._observers[observer].append(filter_)def unregister_observer(self, observer):del self._observers[observer]def notify_observers(self, subject, change):for observer, filter_ in self._observers.items():if filter_ is None or all(filter_.should_notify(observer, subject, change) for filter_ in filter_):observer.update(subject, change)class Subject:def __init__(self):self._mediator = Mediator()def attach(self, observer, filter_=None):self._mediator.register_observer(observer, filter_)def detach(self, observer):self._mediator.unregister_observer(observer)def notify(self, change):self._mediator.notify_observers(self, change)class ConcreteFilter(Filter):def should_notify(self, observer, subject, change):# 根据change内容决定是否通知观察者，这里只是一个简单的示例if change['type'] == 'important':return Truereturn Falseclass Observer(ABC):@abstractmethoddef update(self, subject, change):passclass ConcreteObserver(Observer):def update(self, subject, change):print(f"Observer: Received update: {change}")# 示例使用
subject = Subject()
observer1 = ConcreteObserver()
filter1 = ConcreteFilter()subject.attach(observer1, filter1)
subject.notify({'type': 'normal', 'data': 'Normal update'})
subject.notify({'type': 'important', 'data': 'Important update'})# 输出：
# Observer: Received update: {'type': 'important', 'data': 'Important update'}
在这个示例中，Mediator类替代了原有的直接通知观察者的方式，负责管理观察者列表和过滤器。ConcreteFilter是一个具体的过滤器实现，它定义了何时应该通知观察者。Subject类通过Mediator来注册和通知观察者，并且在通知时通过过滤器进行筛选。这样，只有满足过滤条件的观察者才会接收到通知。
六、总结
在Python中实现观察者模式可以帮助我们更好地组织复杂的业务逻辑，尤其是在处理涉及状态变更和多方响应的情况时。通过合理的应用和适当的优化，可以在多种场景下有效地降低模块之间的耦合度，实现系统内部的良好通信和协调工作。然而，在使用过程中也需要警惕过度使用导致的“扇出”问题（即一个对象的变化可能引发大量其他对象的反应），以及由此带来的性能影响和调试难度的增加。因此，在实际项目中需根据实际情况适度使用观察者模式，并结合其他设计模式来优化整体架构。
关注gzh不灵兔，Python学习不迷路，关注后后台私信，可进wx交流群，进群暗号【人生苦短】~~~
