注册    登录
关注
Py学习  »  Python

一看就懂【来自英雄联盟盖伦的怒吼】与 Python 详解设计模式（二）观察者模式

进击的Coder • 5 年前 • 388 次点击  
“

阅读本文大概需要 8 分钟。


”


观察者模式概述

观察者模式（有时又被称为模型-视图（View）模式、源-收听者(Listener)模式或从属者模式）是软件设计模式的一种。在此种模式中，一个目标物件管理所有相依于它的观察者物件，并且在它本身的状态改变时主动发出通知。这通常透过呼叫各观察者所提供的方法来实现。此种模式通常被用来实现事件处理系统。

基本介绍

观察者模式（Observer）完美的将观察者和被观察的对象分离开。举个例子，用户界面可以作为一个观察者，业务数据是被观察者，用户界面观察业务数据的变化，发现数据变化后，就显示在界面上。面向对象设计的一个原则是：系统中的每个类将重点放在某一个功能上，而不是其他方面。一个对象只做一件事情，并且将他做好。观察者模式在模块之间划定了清晰的界限，提高了应用程序的可维护性和重用性。

观察者设计模式定义了对象间的一种一对多的组合关系，以便一个对象的状态发生变化时，所有依赖于它的对象都得到通知并自动刷新。

实现方式

观察者模式有很多实现方式，从根本上说，该模式必须包含两个角色：观察者和被观察对象。在刚才的例子中，业务数据是被观察对象，用户界面是观察者。观察者和被观察者之间存在“观察”的逻辑关联，当被观察者发生改变的时候，观察者就会观察到这样的变化，并且做出相应的响应。

如果在用户界面、业务数据之间使用这样的观察过程，可以确保界面和数据之间划清界限，假定应用程序的需求发生变化，需要修改界面的表现，只需要重新构建一个用户界面，业务数据不需要发生变化。

观察

实现观察者模式的时候要注意，观察者和被观察对象之间的互动关系不能体现成类之间的直接调用，否则就将使观察者和被观察对象之间紧密的耦合起来，从根本上违反面向对象的设计的原则。无论是观察者“观察”观察对象，还是被观察者将自己的改变“通知”观察者，都不应该直接调用。

观察者模式 UML 图

上面的文字太多了，我们直接看图吧

从图上可以看到，观察者模式主要有 3 个角色：

  • 主题，主题类中有许多的方法，比如 register() 和 deregister() 等，观察者 Observer 可以通过这些方法注册到主题中或从主题注销。一个主题可以对应多个观察者，你可以将它理解为一条消息。

  • 观察者，它为关注主题的对象定义了一个 notify() 接口，以便在主题发生变化时能够获得相应的通知。你可以将它理解为消息推送功能。

  • 具体观察者，它是先了观察者的接口以保持其状态与主题中的变化一致，你可以将它理解为每个英雄，比如德邦总管赵信、德玛西亚皇子嘉文四世、放逐之刃锐雯等，当然了还有迅捷斥候提莫。

这个流程并不复杂，具体观察者（比如嘉文四世、锐雯）通过观察者提供的接口向主题注册自己，每当主题状态发生变化时，该主题都会使用观察者（消息推送功能）提供的通知方式来告知所有的具体观察者（赵信、嘉文、提莫、锐雯）发生了什么。

为什么选择英雄联盟？

因为大家对英雄联盟都熟悉啊，而且这不是 IG 为 LPL 赛区夺得第一个 S 赛冠军了嘛，我正好蹭一波热度。

我可以选择其它游戏么？

可以，只要你能够在你熟悉的领域找到合适的案例来理解，哪怕你用坦克大战来做例子都是可以的。

英雄联盟的通知是什么样的？


你最熟悉的声音莫过于以下几句了：

  • 欢迎来到英雄联盟

  • 敌军还有30秒到达战场，碾碎他们

  • 全军出击

  • First blood

  • Ace
    英雄联盟的消息会在触发事件（比如时间或者某个行为）的时候给部分召唤师或者全部召唤师推送消息。

消息通知的过程

熟悉的台词都可以背得出来了，可你知道这些消息从产生到推给每个召唤师的过程是怎么样的么？

那我们来整理一下顺序吧：

  • 事件触发

  • 产生消息

  • 将消息放到队列

  • 其他召唤师监听队列

  • 队列变化则收到消息

消息通知的过程

思考：这个过程并不复杂，如果根据上方的流程图和顺序，你可以写出消息推送的代码吗？

盖伦的特点是什么

盖伦是英雄联盟中最有特点也最令人映像深刻的角色，一提到他，我们想到的必定是他那超大号的大宝剑和开大招时候那一声 『德玛西亚』的怒吼。

慢着，德玛西亚？

德玛西亚的是如何传到各位召唤师耳朵里的呢？

上面了解了观察者模式的基本，我们心里对代码就会有一个大概的轮廓。比如编写一个消息通知的类、一个消息队列、一个观察者和10个具体观察者（英雄联盟每局10个玩家）。

消息如何传播呢？

消息队列有了，那么如何在触发事件（盖伦开大招）的时候将那一声『德玛西亚』传达到广大英雄(召唤师)的耳朵里呢？

你又如何确定该传到谁那里，但是又要注意排除那些离得远的英雄。

最重要的消息类

首先我们新建一个消息类，这个消息类中需要提供一个供英雄使用的接口，能够让观察者来注册和注销，并且维护一个订阅者队列以及最后一条消息：

class NewsPublisher(object):
    """ 消息主题类 """

    def __init__(self):
        self.__subscribers = []
        self.__latest_news = None

    def register(self, subcriber):
        """ 观察者注册 """
        self.__subscribers.append(subcriber)

    def detach(self):
        """ 观察者注销 """
        return self.__subscribers.pop()

接着还需要什么呢？队列有了，那订阅者列表和负责消息通知的方法还没有，而且消息创建和最新消息的接口也需要编写，那么就将消息类改为：

class NewsPublisher(object):
    """ 消息主题类 """




    
    def __init__(self):
        self.__subscribers = []
        self.__latest_news = None

    def register(self, subcriber):
        """ 观察者注册 """
        self.__subscribers.append(subcriber)

    def detach(self):
        """ 观察者注销 """
        return self.__subscribers.pop()

    def subscribers(self):
        """ 订阅者列表 """
        return [type(x).__name__ for x in self.__subscribers]

    def notify_subscribers(self):
        """ 遍历列表，通知订阅者 """
        for sub in self.__subscribers:
            sub.update()

    def add_news(self, news):
        """ 新增消息 """
        self.__latest_news = news

    def get_news(self):
        """ 获取新消息 """
        return "收到新消息:", self.__latest_news

观察者接口

然后就要考虑观察者接口了，观察者接口是应该是一个抽象基类，具体观察者（英雄）继承观察者。观察者接口需要有一个监听方法，只要有新消息发出，那么所有符合条件的具体观察者就可以收到相应的消息：




    
from abc import ABCMeta, abstractmethod

class Subscriber(metaclass=ABCMeta):
    """ 观察者接口 """
    @ abstractmethod
    def update(self):
        pass

英雄登场


终于到了英雄们盛大登场的时候，所有的英雄的身份在这里都是具体观察者。每个英雄的 init() 方法都通过 register() 方法向消息类进行注册的，你可以理解为在开局画面的时候，就是完成各个英雄之间的类的注册。

英雄也要有一个 update() 方法，以便消息类可以向英雄推送消息：

class Garen(object):
    """ 盖伦 """
    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())


class JarvanIV(object):
    """ 嘉文四世 """

    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())





    
class Riven (object):
    """ 锐雯 """

    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())


class Quinn(object):
    """ 德玛西亚之翼 """

    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())


class XinZhao (object):
    """ 德邦总管 """

    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())


class AurelionSol(object):
    """ 铸星龙王 """

    def __init__(self, publisher):



    
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())


class Aatrox(object):
    """ 暗裔剑魔 """

    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())


class Ryze(object):
    """ 流浪法师 """

    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())


class Teemo(object):
    """ 迅捷斥候 """

    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())


class Malzahar


    
 (object):
    """ 玛尔扎哈 """

    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.register(self)

    def update(self):
        print(type(self).__name__, self.publisher.get_news())

召唤师峡谷现在站着十位英雄，意味着游戏现在开始了。

德玛西亚！


游戏很快就进行了几分钟，现在盖伦升到 6 级了，并升级了 R 技能。盖伦面对上路对线的暗裔剑魔，放出了 Q W E 技能，看见残血的剑魔，盖伦放出了 R 技能。

谁能听到这一声 德玛西亚 ？


玩过的都知道，屏幕视野必须跟说话的英雄在同一屏幕才能听到声音。那么现在我们就为每个英雄都设置一个临是的坐标，并假设在坐标值 200 码内属于同一个屏幕：
if __name__ == "__main__":
    news_publisher = NewsPublisher()  # 实例化消息类
    garen_position = (566, 300)  # 设定盖伦当前位置
    # 各个英雄当前位置
    role_position = [(JarvanIV, 220, 60), (Riven, 56, 235), (Ryze, 1090, 990),
                     (XinZhao, 0, 0), (Teemo, 500, 500), (Malzahar, 69, 200),
                     (Aatrox, 460, 371), (AurelionSol, 908, 2098), (Quinn, 1886, 709)]
def valid_position(role_a: int, role_b: int):



    
    # 同屏幕范围确认
    if abs(role_a - role_b) 200:
        return True
    return False

for sub in role_position:
    if valid_position(sub[1], garen_position[0]) or valid_position(sub[2], garen_position[1]):
        # 只发送给同屏幕的英雄
        sub[0](news_publisher)

改定义的都定义好了，坐标和同屏幕英雄也区分出来了，如何发送新消息呢？

print("在同一个屏幕的英雄有：", news_publisher.subscribers())
news_publisher.add_news("德玛西亚！")
news_publisher.notify_subscribers()

首先通过订阅者列表确认同屏幕的英雄，通过消息类中的 add_news() 方法发出盖伦的怒吼『德玛西亚』，接着使用消息类中的 notify_subscribers()  方法通知订阅者列表中所有英雄。

看看输出结果是什么：

在同一个屏幕的英雄有： ['Riven', 'Teemo', 'Aatrox']
Riven ('收到新消息:', '德玛西亚！')
Teemo ('收到新消息:', '德玛西亚！')
Aatrox ('收到新消息:', '德玛西亚！')

就这样，『德玛西亚』的声音传到了暗裔剑魔、迅捷斥候和放逐之刃那里。

再看一遍

如果第一遍看不懂，可以一边看着 UML 图一边动手实践一遍，就能够彻底理解观察者模式了。

崔庆才

静觅博客博主，《Python3网络爬虫开发实战》作者

隐形字

个人公众号：进击的Coder

长按识别二维码关注




今天看啥 - 高品质阅读平台
本文地址：http://www.jintiankansha.me/t/ueHqsaiWC8
Python社区是高质量的Python/Django开发社区
本文地址：http://www.python88.com/topic/26353
Reply
 
388 次点击  
登录后回复
关于   移动版   ·   三行代码   ·   今天看啥   ·   Code   ·   link之家   ·   卧龙搜索   ·   藏经阁   ·   小百科
Py学习 - 专注于Python技术发展的社区(原Django社区)
沪ICP备11025650号