Python 实现观察者模式：发布订阅机制详解

核心概念与机制

观察者模式主要包含两个角色：主题（Publisher）和观察者（Subscriber）。主题负责维护一个观察者列表，并在自身状态发生变化时，主动通知所有注册的观察者。这种机制确保了主题与观察者之间的松耦合关系，使得系统可以灵活地添加或删除观察者，而无需修改核心业务逻辑。

该模式支持两种数据传递方式：

• 拉模型（Pull）：观察者收到通知后，主动向主题请求最新数据。
• 推模型（Push）：主题在通知时直接推送相关数据给观察者。

通过这种设计，一个主题可以对应多个观察者，且可以在不修改主题或观察者代码的前提下，高效地扩展数据发送功能。

Python 代码实现

在 Python 中，我们可以利用 abc 模块中的抽象基类来定义观察者的标准接口，确保所有具体观察者都实现必要的方法。以下是一个完整的发布订阅示例：

from abc import ABCMeta, abstractmethod


class Publisher:
  """被观察者：发布/订阅关系中的发布对象"""
  def __init__(self):
    self.subscribers = []
    self.latest_content = None

  def set_content(self, content):
    """有新消息时，发布新的消息"""
    self.latest_content = content
    self.publish()

  def get_latest_content(self):
    """获取最新的消息"""
    return self.latest_content

  def register(self, subscriber):
    """注册一个新的订阅者"""
    self.subscribers.append(subscriber)

  def publish(self):
    """发布消息并通知订阅的用户"""
    for subscriber in self.subscribers:
      subscriber.notify()


class Subscriber(metaclass=ABCMeta):
  """观察者的抽象类：需要定义一个通知接口，用于发布对象通知订阅的用户"""
  @abstractmethod
  def notify(self):
    pass


class SubscriberA(Subscriber):
  """观察者A：发布/订阅关系中的订阅者，当订阅的发布者有新的变化或动态的时候能及时收到通知"""
  def __init__(self):
    self.my_publisher = None

  def subscribe(self, publisher):
    """订阅并进行注册"""
    self.my_publisher = publisher
    self.my_publisher.register(self)

  def notify(self):
    """获取最新消息"""
    latest_content = self.my_publisher.get_latest_content()
    print(self, latest_content)


class SubscriberB(Subscriber):
  """观察者B：发布/订阅关系中的订阅者，当订阅的发布者有新的变化或动态的时候能及时收到通知"""
  def __init__(self):
    self.my_publisher = None

  def subscribe(self, publisher):
    """订阅并进行注册"""
    self.my_publisher = publisher
    self.my_publisher.register(self)

  def notify(self):
    """获取最新消息"""
    latest_content = self.my_publisher.get_latest_content()
    print(self, latest_content)


if __name__ == '__main__':
  new_publisher = Publisher()
  subscriber_a = SubscriberA()
  subscriber_a.subscribe(new_publisher)
  subscriber_b = SubscriberB()
  subscriber_b.subscribe(new_publisher)
  new_publisher.set_content('This is a new message!')

代码逻辑解析

上述代码展示了观察者模式的典型实现流程：

1. 定义发布者（Publisher）：包含一个订阅者列表 subscribers 和最新内容 latest_content。提供 register 方法用于添加观察者，set_content 用于更新状态并触发 publish 方法。
2. 定义抽象观察者（Subscriber）：使用 ABCMeta 元类，强制子类实现 notify 方法。
3. 具体观察者（SubscriberA/B）：继承自 Subscriber，在 subscribe 方法中自动完成注册操作。在 notify 方法中，通过拉模型获取最新内容并打印。
4. 运行测试：创建发布者实例和两个观察者实例，分别订阅后，更新发布者内容，触发所有观察者的通知。

通过这种方式，系统实现了高度的解耦。发布者无需知道观察者的具体实现，只需调用其 notify 方法；观察者也无需关心发布者的内部状态，只需实现通知逻辑即可。

以上就是 Python 实现观察者模式的详细内容，更多关于 Python 设计模式与解耦技术的资料请关注本站其它相关文章！