1. 程式人生 > >一看就懂【來自英雄聯盟蓋倫的怒吼】與 Python 詳解設計模式(二)觀察者模式

一看就懂【來自英雄聯盟蓋倫的怒吼】與 Python 詳解設計模式(二)觀察者模式

觀察者模式概述

觀察者模式(有時又被稱為模型-檢視(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 圖一邊動手實踐一遍,就能夠徹底理解觀察者模式了。