Python观察者模式介绍、使用方法-Toy模板网

这篇具有很好参考价值的文章主要介绍了Python观察者模式介绍、使用方法。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一、Python观察者模式介绍

Python观察者模式是一种行为型设计模式，它将对象分成两个部分：观察者和主题。观察者在主题状态改变时被通知并且更新自己。

功能：

通过解耦来实现可重用性和灵活性。
提供了一种对象间的一对多依赖关系，当一个对象改变状态时，所有依赖对象都会收到通知。
主题和观察者之间是松散耦合的，它们可以独立地改变。

优点：

提供灵活性，减少类间的耦合度。
可以实现动态的发布-订阅机制，并且观察者可在运行时动态地加入或离开主题。
实现开放-封闭原则，使得主题和观察者可以独立地扩展和修改。

缺点：

观察者过多会导致通知效率等性能问题。
如果观察者与主题相互依赖，修改其中任意一方的代码都需要修改对方的代码，不利于系统的扩展和维护。

应用场景：

当一个对象的改变需要同时改变其他对象时，比如消息订阅模型。
当一个抽象模型有两个方面，其中一个方面需要依赖另外一个方面时。

使用方式：

定义观察者类，包含需要执行的方法。
定义主题类，包含观察者列表和通知观察者的方法。
在主题类的方法中通知观察者。

在应用程序开发中的应用：

GUI界面中的控件更新。
消息订阅模型。
日志记录器。

二、观察者模式使用

工作原理：

观察者将自己注册到主题中。
主题维护一个观察者列表。
当主题状态发生改变时，通知所有观察者。
观察者执行相应的操作。

示例一：实现信息通知功能

假设我们要实现一个天气预报系统，当天气发生变化时，需要将信息通知给所有的观察者。这是一个常见的观察者模式的应用场景，下面举例详细说明 Python观察者模式的工作原理和使用方法。

首先，我们需要定义两个类：主题类和观察者类。主题类包含一个观察者列表和通知观察者的方法，观察者类包含需要执行的方法。WeatherSubject类为天气主题类，包含observers列表和register_observer、remove_observer、notify_observers、set_temperature等方法。WeatherObserver类为观察者类，包含name属性和update方法。在update方法中，观察者会执行相应的操作，例如打印收到通知的信息。

接下来，我们可以创建主题对象和观察者对象，并将观察者对象注册到主题对象中。创建了一个主题对象subject和两个观察者对象observer_1和observer_2，并将观察者对象注册到主题对象中。

现在，我们可以改变天气信息，然后通知所有的观察者。代码如下：

# *********************************观察者模式实现信息通知功能
# 主题类
class WeatherSubject():
    def __init__(self):
        self.observers = [] # 初始化观察者数组
        self.temperature = None # 初始化温度
    def register_observer(self, observer):
        self.observers.append(observer)

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

    def notify_observer(self):
        for observer in self.observers:
            observer.update(self.temperature) # 通知观察者温度

    def set_temperature(self, temperature):
        self.temperature = temperature # 设置温度
        self.notify_observer() # 通知观察者

# 观察者类
class WeatherObserver():
    def __init__(self, name):
        self.name = name
    def update(self, temperature):
        print(f"{self.name}收到温度通知：温度为{temperature}℃")

subject = WeatherSubject()

observer1 = WeatherObserver("name1")
observer2 = WeatherObserver("name2")

subject.register_observer(observer1)
subject.register_observer(observer2)

subject.set_temperature(20)

运行结果

name1收到温度通知：温度为20℃
name2收到温度通知：温度为20℃

在上述代码中，我们将温度设置为20℃，然后调用set_temperature方法通知所有的观察者。观察者会收到通知并执行相应的操作，例如打印收到通知的信息。

这就是 Python观察者模式的工作原理和使用方法。当主题状态发生变化时，通知所有的观察者，执行相应的操作，实现了对象间的一对多依赖关系。

示例二：实现消息订阅功能

下面我们以一个简单的消息订阅系统为例，详细说明 Python观察者模式实现消息订阅功能。

首先，我们需要定义两个类：主题类和观察者类。主题类包含一个观察者列表和通知观察者的方法，观察者类包含需要执行的方法。，NewsSubject类为新闻主题类，包含observers字典和register_observer、remove_observer、notify_observers、set_news等方法。其中，observers字典的键为订阅主题，值为观察者列表。NewsObserver类为观察者类，包含name属性和update方法。

接下来，我们可以创建主题对象和观察者对象，并将观察者对象注册到主题对象中。创建了一个主题对象subject和三个观察者对象observer_1、observer_2和observer_3，并将观察者对象注册到主题对象中，每个观察者订阅不同的新闻主题。

现在，我们可以发布新闻消息，然后通知订阅相应主题的观察者。具体代码如下：

# *********************************观察者模式实现消息订阅功能

# 新闻主题类
class NewsSubject():
    def __init__(self):
        self.observers = {} # 字典
        self.news = None

    def register_observer(self, observer, topic):
        if topic not in self.observers:
            self.observers[topic] = []
        self.observers[topic].append(observer)

    def remove_observer(self, observer, topic):
        if topic in self.observers:
            self.observers[topic].remove(observer)

    def notify_observers(self,topic):
        if topic in self.observers: # if语句，存在观察者订阅的主题新闻，通知观察者
            for observer in self.observers[topic]:
                observer.update(self.news)  # 通知观察者新闻

    def set_news(self, news, topic):
        self.news = news
        self.notify_observers(topic)



class NewsObserver():
    def __init__(self, name):
        self.name = name

    def update(self, news):
        print(f"{self.name}收到新闻：{news}")

subject = NewsSubject()

observer_1 = NewsObserver("name1")
observer_2 = NewsObserver("name2")
observer_3 = NewsObserver("name3")

subject.register_observer(observer_1, "国内新闻")
subject.register_observer(observer_2, "国外新闻")
subject.register_observer(observer_3, "娱乐新闻")

subject.set_news("中国的新闻，中国......", "国内新闻")
subject.set_news("外国的新闻，美国......", "国外新闻")
subject.set_news("明星...","娱乐新闻")
subject.set_news("歌星...","娱乐新闻")

运行结果：

name1收到新闻：中国的新闻，中国......
name2收到新闻：外国的新闻，美国......
name3收到新闻：明星...
name3收到新闻：歌星...

在上述代码中，我们发布了三条新闻消息，并使用set_news方法通知订阅相应主题的观察者。观察者会收到通知并执行相应的操作，例如打印收到通知的信息。

这就是 Python观察者模式实现消息订阅功能的方法。当主题状态发生变化时，通知所有订阅该主题的观察者，执行相应的操作，实现了消息订阅功能。

示例三：实现日志记录器功能

下面我们以一个简单的日志记录器系统为例，详细说明 Python观察者模式实现日志记录器功能。

首先，我们需要定义两个类：主题类和观察者类。主题类包含一个观察者列表和通知观察者的方法，观察者类包含需要执行的方法。LoggerSubject类为日志主题类，包含observers字典和register_observer、remove_observer、notify_observers等方法。其中，observers字典的键为日志级别，值为观察者列表。FileLoggerObserver类为观察者类，包含log_file属性和update方法。在update方法中，我们将收到的日志消息写入指定的日志文件中。

接下来，我们可以创建主题对象和观察者对象，并将观察者对象注册到主题对象中。创建了一个主题对象subject和三个观察者对象observer_1、observer_2和observer_3，并将观察者对象注册到主题对象中，每个观察者订阅不同的日志级别。

现在，我们可以记录日志消息，然后通知订阅相应级别的观察者。具体代码如下：

# *********************************观察者模式实现日志记录功能

# 日志主题类
class LoggerSubject():
    def __init__(self):
        self.observers = {} # 字典: 日志级别

    def register_observer(self, observer, level):
        if level not in self.observers:
            self.observers[level] = []
        self.observers[level].append(observer)

    def remove_observer(self, observer, level):
        if level in self.observers:
            self.observers[level].remove(observer)

    def notify_observers(self,message,level):
        if level in self.observers: # if语句，存在观察者订阅的主题新闻，通知观察者
            for observer in self.observers[level]:
                observer.update(message)  # 通知观察者新闻

# 观察者类
class FileLoggerObserver():
    def __init__(self, log_file):
        self.log_file = log_file

    def update(self, message):
        print(f"{self.log_file}写入日志：{message}")
        with open(self.log_file, "a") as f:
            f.write(message + "\n")

subject = LoggerSubject()

observer_1 = FileLoggerObserver("log_debug.txt")
observer_2 = FileLoggerObserver("log_info.txt")
observer_3 = FileLoggerObserver("log_error.txt")

subject.register_observer(observer_1,"debug")
subject.register_observer(observer_2, "info")
subject.register_observer(observer_3, "error")

subject.notify_observers("调试信息","debug")
subject.notify_observers("普通信息","info")
subject.notify_observers("错误信息","error")
subject.notify_observers("错误信息1","error")

运行结果：