当前位置：首页 > news >正文

设计模式Python版观察者模式

news 来源：原创 2025/5/26 9:21:07

文章目录

前言
一、观察者模式
二、观察者模式示例

前言

GOF设计模式分三大类：

创建型模式：关注对象的创建过程，包括单例模式、简单工厂模式、工厂方法模式、抽象工厂模式、原型模式和建造者模式。
结构型模式：关注类和对象之间的组合，包括适配器模式、桥接模式、组合模式、装饰模式、外观模式、享元模式和代理模式。
行为型模式：关注对象之间的交互，包括职责链模式、命令模式、解释器模式、迭代器模式、中介者模式、备忘录模式、观察者模式、状态模式、策略模式、模板方法模式和访问者模式。

一、观察者模式

观察者模式（Observer Pattern）

定义：定义对象之间的一种一对多依赖关系，使得每当一个对象状态发生改变时，其相关依赖对象皆得到通知并被自动更新。观察者模式的别名包括发布-订阅（Publish/Subscribe）模式、模型-视图（Model/View）模式、源-监听器（Source/Listener）模式或从属者（Dependents）模式。
解决问题：如何实现对象之间一对多的联动？
使用场景：
- 一个对象的改变将导致一个或多个其他对象也发生改变，而并不知道具体有多少对象将发生改变，也不知道这些对象是谁。
- 需要在系统中创建一个触发链，A对象的行为将影响B对象，B对象的行为将影响C对象……
组成：
- Subject（目标）：目标又称为主题，它是指被观察的对象。在目标中定义了一个观察者集合，一个观察目标可以接受任意数量的观察者来观察，它提供一系列方法来增加和删除观察者对象，同时定义了通知方法notify()
- ConcreteSubject（具体目标）：具体目标是目标类的子类，通常包含有经常发生改变的数据。当它的状态发生改变时，向其各个观察者发出通知。
- Observer（观察者）：观察者将对观察目标的改变做出反应。观察者一般定义为接口，该接口声明了更新数据的方法update（），因此又称为抽象观察者。
- ConcreteObserver（具体观察者）：在具体观察者中维护一个指向具体目标对象的引用，它存储具体观察者的有关状态，这些状态需要和具体目标的状态保持一致。
补充说明：
- 一个对象的状态或行为的变化将导致其他对象的状态或行为也发生改变，它们之间将产生联动，正所谓“触一而牵百发”。
- 在观察者模式中，发生改变的对象称为观察目标，而被通知的对象称为观察者。
- 一个观察目标可以对应多个观察者，而且这些观察者之间可以没有任何相互联系，可以根据需要增加和删除观察者，使得系统更易于扩展。
- 观察者模式是一种使用频率非常高的设计模式
优点：
- 观察者模式可以实现表示层和数据逻辑层的分离。
- 观察者模式在观察目标和观察者之间建立一个抽象的耦合。
- 观察者模式支持广播通信。
- 观察者模式满足开闭原则的要求，增加新的具体观察者无须修改原有系统代码。
缺点：
- 将所有的观察者都通知到会花费很多时间
- 如果在观察者和观察目标之间存在循环依赖，可能导致系统崩溃
- 仅仅只是知道观察目标发生了变化，但不知道具体变化

在这里插入图片描述

二、观察者模式示例

使用观察者模式来进行多人联机对战游戏的设计

多个玩家可以加入同一战队组成联盟，当战队中某一成员受到敌人攻击时将给所有其他盟友发送通知，盟友收到通知后将做出响应。
AllyControlCenter充当目标类，ConcreteAllyControlCenter充当具体目标类，Observer充当抽象观察者，Player充当具体观察者。
实现了两次对象之间的联动：当一个游戏玩家Player对象的beAttacked（）方法被调用时，将调用AllyControlCenter的notifyObserver（）方法来进行处理，而在notifyObserver（）方法中又将调用其他Player对象的help（）方法。
- Player.beAttacked() → AllyControlCenter.notifyObserver() → Player.help()

"""观察者模式"""

### 目标类


class AllyControlCenter:
    """战队控制中心"""

    def __init__(self, ally_name):
        self.ally_name = ally_name  # 战队名称
        self.players: list[Observer] = []  # 用于存储战队成员

    def join(self, obs: "Observer"):
        """注册方法"""
        print(f"{obs.name} 加入 {self.ally_name} 战队！")
        self.players.append(obs)

    def quit(self, obs: "Observer"):
        """注销方法"""
        print(f"{obs.name} 退出 {self.ally_name} 战队！")
        self.players.remove(obs)

    def notify_observer(self, name: str):
        """通知方法"""
        raise NotImplementedError


### 具体目标类


class ConcreteAllyControlCenter(AllyControlCenter):
    """具体战队控制中心"""

    def __init__(self, ally_name):
        super().__init__(ally_name)
        print(f"{self.ally_name} 战队组建成功！")
        print("#" * 10)

    def notify_observer(self, name):
        print(f"{self.ally_name} 战队紧急通知，盟友 {name} 遭受敌人攻击！")
        # 遍历调用每一个盟友（自己除外）的支援方法
        for obs in self.players:
            if obs.name != name:
                obs.help()


### 抽象观察者


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

    def help(self):
        """支援盟友方法"""
        raise NotImplementedError

    def be_attacked(self, acc: AllyControlCenter):
        """遭受攻击方法"""
        raise NotImplementedError


### 具体观察者


class Player(Observer):
    """战队成员"""

    def help(self):
        print(f"坚持住，{self.name} 来支援你！")

    def be_attacked(self, acc):
        """当遭受攻击时将调用战队控制中心类的通知方法来通知盟友"""
        print(f"{self.name} 被攻击！")
        acc.notify_observer(self.name)

客户端代码

if __name__ == "__main__":
    acc = ConcreteAllyControlCenter("金陵十二钗")

    player1 = Player("林黛玉")
    acc.join(player1)
    player2 = Player("薛宝钗")
    acc.join(player2)
    player3 = Player("贾元春")
    acc.join(player3)
    player4 = Player("贾探春")
    acc.join(player4)

    player1.be_attacked(acc)

输出结果

金陵十二钗 战队组建成功！
##########
林黛玉 加入 金陵十二钗 战队！
薛宝钗 加入 金陵十二钗 战队！
贾元春 加入 金陵十二钗 战队！
贾探春 加入 金陵十二钗 战队！
林黛玉 被攻击！
金陵十二钗 战队紧急通知，盟友 林黛玉 遭受敌人攻击！
坚持住，薛宝钗 来支援你！
坚持住，贾元春 来支援你！
坚持住，贾探春 来支援你！

您正在阅读的是《设计模式Python版》专栏！关注不迷路~