Python设计模式之观察者模式简单示例

半杯拿铁

发布时间：2026-10-13| 2人浏览过

原创

本文通过Python代码详解观察者模式，以TechForum技术论坛为例，展示发布者与订阅者的交互机制。内容涵盖核心概念、类结构分析及完整实现代码，帮助开发者理解分布式事件处理中的状态同步与通知机制。

本文通过具体实例解析Python中的观察者设计模式，旨在为读者提供清晰的参考与学习素材。

观察者模式属于软件设计模式的一种。其核心机制在于：一个主题对象维护着一系列依赖它的观察者列表，当主题状态发生改变时，会自动通知所有观察者，通常通过调用观察者的特定方法来实现。该模式特别适用于分布式事件处理系统。

在典型的观察者模式架构中，主要包含以下要素：

1. 发布者（Publisher）类需具备以下功能：

注册接收通知的对象；从主题向已注册对象发送变更通知；确保未注册的对象无法接收任何通知信息。

2. 订阅者（Subscriber）类需具备以下特征：

发布者会调用订阅者提供的方法，将任何状态变更告知已注册的订阅对象。

3. 当触发事件导致状态改变时，发布者将调用通知方法。

总结而言，订阅者可以选择在发布对象中注册或不注册。无论发生何种事件，发布者都会通过调用通知方法来告知订阅者。但请注意，这种通知仅在事件发生时，针对已注册的订阅者生效。

以下是一个简单的Python实现示例：

我们构建一个TechForum技术论坛场景，模拟不同用户发布技术邮件的过程。当任何用户发布新邮件时，其他用户将收到新邮件的通知。从对象设计的角度来看，我们需要一个TechForum对象作为发布者，并允许其他用户对象在TechForum上注册，以便在新邮件通知时接收邮件标题。

这个简单的例子可以类比为中介机构与雇主、求职者之间的关系。这延伸了招聘者与应聘者的关系：通过工作中介发布各类工作信息，求职者寻找相关工作，招聘者则在中介处寻找注册的求职者。

具体实现代码如下：

class Publisher:
  def __init__(self):
    pass
  def register(self):
    pass
  def unregister(self):
    pass
  def notifyAll(self):
    pass
class TechForum(Publisher):
  def __init__(self):
    self._listOfUsers = []
    self.postname = None
  def register(self, userObj):
    if userObj not in self._listOfUsers:
      self._listOfUsers.append(userObj)
  def unregister(self, userObj):
    self._listOfUsers.remove(userObj)
  def notifyAll(self):
    for objects in self._listOfUsers:
      objects.notify(self.postname)
  def writeNewPost(self , postname):
    self.postname = postname
    self.notifyAll()
class Subscriber:
  def __init__(self):
    pass
  def notify(self):
    pass
class User1(Subscriber):
  def notify(self, postname):
    print "User1 notified of a new post %s" % postname
class User2(Subscriber):
  def notify(self, postname):
    print "User2 notified of a new post %s" % postname
class SisterSites(Subscriber):
  def __init__(self):
    self._sisterWebsites = ["Site1" , "Site2", "Site3"]
  def notify(self, postname):
    for site in self._sisterWebsites:
        print "Send nofication to site:%s " % site
if __name__ == "__main__":
  techForum = TechForum()
  user1 = User1()
  user2 = User2()
  sites = SisterSites()
  techForum.register(user1)
  techForum.register(user2)
  techForum.register(sites)
  techForum.writeNewPost("Observe Pattern in Python")
  techForum.unregister(user2)
  techForum.writeNewPost("MVC Pattern in Python")

程序运行结果如下：