你是否曾经好奇过，当你最喜欢的频道上传新视频时，YouTube是如何通知你的？或者当有新邮件到达时，你的电子邮件客户端又是如何提醒你的？这些都是观察者模式在实际应用中的绝佳例子。
观察者模式是一种设计模式，在这种模式下，一个对象（称为主题对象）会维护一份依赖对象的列表（这些依赖对象被称为观察者），并且当它的状态发生变化时，会自动通知这些观察者。这就像订阅了一份新闻通讯：每当有新内容发布时，所有订阅者都会收到通知。
在本教程中，你将了解什么是观察者模式、它为什么有用，以及如何通过实际例子在Python中实现它。
你可以在GitHub上找到相关代码。
先决条件
在开始之前，请确保你满足以下要求：
-
已安装Python 3.10或更高版本
-
了解Python类和方法的工作原理
-
熟悉面向对象编程的概念
让我们开始吧！
目录
什么是观察者模式？
观察者模式定义了对象之间的一种一对多关系。当某个对象的状态发生变化时，所有依赖于它的对象都会被自动通知并更新。
可以把这想象成一家新闻机构和它的记者们：当有突发新闻发生时，该机构会立即通知所有订阅了该新闻的记者们；然后每位记者都可以用自己的方式来处理这些新闻——有的人可能会在推特上发布消息，有的人会写文章，还有人会在电视上报道这件事。
以下情况下，观察者模式非常有用：
-
当你需要通知多个对象它们的状态发生了变化时
-
当你希望对象之间的耦合关系较为松散时
-
当你事先不知道到底有多少对象需要被通知时
-
当你希望对象能够动态地订阅或取消订阅时
一个简单的观察者模式示例
让我们从一个基本的例子开始：一个会在有新文章发布时通知读者的博客。
我们将创建一个博客（主题）以及一些电子邮件订阅者（观察者），当有新内容发布时，这些订阅者会自动收到通知。
首先，我们来构建Blog类，这个类将负责管理订阅者并发送通知：
class Blog:
def __init__(self, name):
self.name = name
self._subscribers = []
self._latest_post = None
def subscribe(self, subscriber):
"""将订阅者添加到博客中"""
if subscriber in self._subscribers:
self._subscribers.append(subscriber)
print(f"✓ {self.name}")
def unsubscribe(self, subscriber):
"""将订阅者从博客中移除"""
if subscriber f"✗ {self.name}")
def notify_all(self):
"""向所有订阅者发送通知"""
print(f"\n正在通知)
for subscriber def publish_post(self, title):
"""发布新文章并通知订阅者"""
print(f"\n📝 {title}'")
self._latest_post = title
self.notify_all()
Blog类是我们的主题对象。它会在_subscribers中维护一份订阅者列表，并将最新的文章标题存储在_latest_post中。subscribe方法会将新的订阅者添加到列表中，同时会检查是否存在重复的订阅者。notify_all方法则会遍历所有订阅者，并调用他们的receive_notification方法。当我们调用publish_post方法时，系统会更新最新的文章内容，并自动通知所有的订阅者。
现在让我们创建一个用于接收通知的观察者类：
class EmailSubscriber:
def __init__(self, email):
self.email = email
def receive_notification(self, blog_name, post_title):
print(f"已向 {blog_name} 上发布了新文章——标题为 ')
EmailSubscriber类就是我们的观察者。它只有一个方法，即receive_notification，该方法用于处理来自博客的通知。
现在让我们将这些类结合起来使用：
# 创建一个博客
tech_blog = Blog("DevDaily")
# 创建订阅者
reader1 = EmailSubscriber("anna@example.com")
reader2 = EmailSubscriber("betty@example.com")
reader3 = EmailSubscriber("cathy@example.com")
# 订阅该博客
tech_blog.subscribe(reader1)
tech_blog.subscribe(reader2)
tech_blog.subscribe(reader3)
# 发布文章
tech_blog.publish_post("10条面向初学者的Python技巧")
tech_blog.publish_post("理解设计模式")
输出结果：
✓ anna@example.com 已订阅DevDaily
✓ betty@example.com 已订阅DevDaily
✓ cathy@example.com 已订阅DevDaily
📝 DevDaily发布了新文章：'10条面向初学者的Python技巧'
正在通知3位订阅者…
📧 已向 anna@example.com 发送电子邮件：DevDaily发布了新文章——标题为 '10条面向初学者的Python技巧'
📧 已向 betty@example.com 发送电子邮件：DevDaily发布了新文章——标题为 '10条面向初学者的Python技巧'
📧 已向 cathy@example.com 发送电子邮件：DevDaily发布了新文章——标题为 '10条面向初学者的Python技巧'
📝 DevDaily发布了新文章：'理解设计模式'
正在通知3位订阅者…
📧 已向 anna@example.com 发送电子邮件：DevDaily发布了新文章——标题为 '理解设计模式'
📧 已向 betty@example.com 发送电子邮件：DevDaily发布了新文章——标题为 '理解设计模式'
📧 已向 cathy@example.com 发送电子邮件：DevDaily发布了新文章——标题为 '理解设计模式'
请注意，Blog类并不需要了解每个订阅者是如何处理通知的细节的；它只需要调用他们的receive_notification方法即可。
注意：可以将这里所有的示例都视为解释观察者模式工作原理的占位函数。在您自己的项目中，会有一些用于连接电子邮件服务或其他系统的函数。
处理取消订阅的操作
在实际应用中，用户需要能够取消订阅。其实现方式如下：
blog = Blog("CodeMaster")
user1 = EmailSubscriber("john@example.com")
user2 = EmailSubscriber("jane@example.com")
# 用户进行订阅
blog.subscribe(user1)
blog.subscribe(user2)
# 发布一篇帖子
blog.publish_post("Getting Started with Python")
# user1取消订阅
blog.unsubscribe(user1)
# 再发布一篇帖子——只有user2会收到通知
blog.publish_post("Advanced Python Techniques")
输出结果如下：
✓ john@example.com 已订阅 CodeMaster
✓ jane@example.com 已订阅 CodeMaster
📝 CodeMaster 发布了新帖子：'Getting Started with Python'
正在通知 2 名订阅者…
📧 向 john@example.com 发送邮件：CodeMaster 的新帖子 - 'Getting Started with Python'
📧 向 jane@example.com 发送邮件：CodeMaster 的新帖子 - 'Getting Started with Python'
✗ john@example.com 已取消订阅 CodeMaster
📝 CodeMaster 发布了新帖子：'Advanced Python Techniques'
正在通知 1 名订阅者…
📧 向 jane@example.com 发送邮件：CodeMaster 的新帖子 - 'Advanced Python Techniques'
在user1取消订阅后，只有user2会收到第二篇帖子的通知。观察者模式使得动态添加或删除观察者变得非常方便。
不同类型的观察者
观察者模式的一个非常实用的特点是：不同的观察者可以对同一事件做出不同的反应。让我们来创建一个股票价格跟踪系统，其中多种类型的观察者会对价格变化作出响应。
首先，我们来创建Stock类，当股票价格发生变化时，该类会通知相应的观察者：
class Stock:
def __init__(self, symbol, price):
self.symbol = symbol
self._price = price
self._observers = []
def add(observer(self, observer):
self._observers.append(observer)
print(f"观察者已添加：{observer.__class__.__name__}")
def remove.Observer(self, observer):
self._observers.remove(observer)
def notify_observers(self):
for observer in self._observers:
observer.update(self.symbol, self._price)
def set_price(self, price):
print(f"\n {self.symbol} 的价格已更改：${self._price} → ${price}")
self._price = price
self.notify_observers()
Stock类会维护当前的价格，每当调用set_price方法时，就会通知所有观察者。
现在让我们创建三种不同类型的观察者，它们会对价格变化做出不同的反应：
class EmailAlert:
def __init__(self, email):
self.email = email
def update(self, symbol, price):
print(f"正在给{symbol}的价格现已变为$)
class SMSAlert:
def __init__(self, phone):
self.phone = phone
def update(self, symbol, price):
print(f>正在给{symbol}的价格现已变为$)
class Logger:
def update(self, symbol, price):
print(f>记录日志：系统时间下，{price}")
每种观察者对`update`方法的实现都是不同的：EmailAlert会发送电子邮件，SMSAlert会发送短信，而Logger则会记录这一变化。
现在让我们将它们一起使用：
# 创建一只股票对象
apple_stock = Stock("AAPL", 150.00)
# 创建不同类型的观察者对象
email notifier = EmailAlert("investor@example.com")
smsNotifier = SMSAlert "+1234567890")
price_logger = Logger()
# 将所有观察者对象添加到股票对象中
apple_stock.add_observer(email_notifier)
apple_stock.add.Observer(sms notifier)
apple_stock.add(observer(price_logger)
# 更新股票价格
apple_stock.set_price(155.50)
apple_stock.set_price(152.25)
输出结果：
已添加观察者：EmailAlert
已添加观察者：SMSAlert
已添加观察者：Logger
AAPL价格发生变化：150.0美元 → 155.5美元
📧 向investor@example.com发送电子邮件：AAPL当前价格为155.5美元
📱 向+1234567890发送短信：AAPL价格更新为155.5美元
📝 日志记录：系统时间下AAPL的价格为155.5美元
AAPL价格发生变化：155.5美元 → 152.25美元
📧 向investor@example.com发送电子邮件：AAPL当前价格为152.25美元
📱 向+1234567890发送短信：AAPL价格更新为152.25美元
📝 日志记录：系统时间下AAPL的价格为152.25美元
Stock类并不关心各个观察者具体执行了什么操作。它只是会调用每个观察者的update方法，并传递必要的数据。你可以随意组合和使用这些观察者。
使用抽象基类
为了确保所有观察者都能遵循统一的接口规范，我们可以利用Python的抽象基类功能。这样就能保证类型安全。
首先，让我们创建那些定义接口的基类：
from abc import ABC, abstractmethod
class Subject(ABC):
def __init__(self):
self._observers = []
def attach(self, observer):
not def detach(self, observer):
self._observers.remove(observer)
def notify(self, data):
in self._observers:
observer.update(data)
class Observer(ABC):
@abstractmethod
def update(self, data):
pass
Subject类提供了标准化的观察者管理方法。Observer类则通过@abstractmethod装饰器定义了接口，确保所有观察者都必须实现update方法。
现在，让我们创建一个使用这些基类的订单系统：
class OrderSystem(Subject):
def __init__(self):
super().__init__()
self._order_id = None
def place_order(self, order_id, items):
print(f"\n🛒 已下达订单 #{len(items)} 件")
self._order_id = order_id
self.notify({"order_id": order_id, "items": items})
OrderSystem 继承自 Subject，因此它可以在不自行实现相关逻辑的情况下管理观察者。
接下来，让我们为不同的部门创建具体的观察者类：
class InventoryObserver(Observer):
def update(self, data):
print(f"📦 库存管理：正在更新订单 #'order_id']} 的库存信息")
class ShippingObserver(Observer):
def update(self, data):
print(f"🚚 运输安排：正在为订单 #'order_id']} 准备发货")
class BillingObserver(Observer):
def update(self, data):
print(f"💳 结账流程：正在处理订单 #'order_id']> 的付款事宜")
每个观察者类都必须实现 update 方法。现在，让我们把所有这些部分组合起来：
# 创建订单系统
order_system = OrderSystem()
# 创建观察者对象
inventory = InventoryObserver()
shipping = ShippingObserver()
billing = BillingObserver()
# 将观察者对象关联到订单系统中
order_system.attach(inventory)
order_system.attach(shipping)
order_system.attach(billing)
# 下单
order_system.place_order("ORD-12345", ["笔记本电脑", "鼠标", "键盘"]))
输出结果：
🛒 已下达订单#ORD-12345，共包含3件商品
📦 库存管理：正在更新订单#ORD-12345的库存信息
🚚 运输安排：正在为订单#ORD-12345准备发货事宜
💳 结账流程：正在处理订单#ORD-12345的付款手续使用抽象基类能够确保类型安全性，并保证所有观察者都遵循相同的接口规范。
何时使用观察者模式
观察者模式适用于以下场景：
-
事件驱动型系统——例如GUI框架、游戏引擎，或是任何需要通过操作触发其他部分更新的系统。
-
实时通知功能——比如聊天应用、社交媒体信息流、股票行情界面或推送通知系统。
-
解耦架构设计——当你希望主体对象与其观察者保持独立关系，以提高系统的灵活性时。
-
多监听器场景——当多个对象需要对同一事件做出不同反应时。
然而，在存在简单的一对一关系时，或者当观察者数量较多且性能成为关键因素时，应避免使用观察者模式（因为通知机制可能会带来较大的性能开销）。
总结
观察者模式有效地实现了产生事件的对象与响应这些事件的对象之间的解耦。它促进了代码的松耦合设计——主体对象无需了解观察者的具体实现细节，只需知道它们具备相应的更新方法即可。
我们介绍了观察者模式的基本实现方式、订阅机制、不同类型的观察者以及抽象基类的应用。建议先从最基础的主体-观察者关系开始学习，只有在确实需要时再逐步增加复杂性。
祝编程愉快！