设计模式 with Python2：观察者模式

2021年6月13日 1152点热度 0人点赞 0条评论

虽然这个系列上一篇已经是在一个多月之前了：设计模式 with Python1：策略模式，但经过系统学习完《Fluent Python》，对Python这个语言的特性有了更深一层的认识，而且对于设计模式也是，就像《Fluent Python》的作者说的那样，设计模式虽然是一种编程语言无关的通用思想，但是具体到某种语言的实现本身，还是会无可避免地因为语言特性而变得天差地别（关于这点，在Python学习笔记23：Python设计模式中有详细说明。）。

所以，在接下来的本系列文章的撰写中，我会结合《Head First 设计模式》一书，以经典的设计模式为主，先阐述经典设计模式的实现思路，然后再探索Python中实现的特点。所以如果经典的实现代码太过啰嗦，不够Python，请不要惊讶，其目的只是使用Python阐明经典设计模式的思路，而非实际使用中的实现案例。

好了，废话不多说了，直接进入观察者模式。

概念

观察者模式本质上是为了解决信息更新的问题，这就像是我平时使用RSS客户端或者查看我的CSDN留言，最简单和最常见的方式当然是轮询，即每过一段时间就去刷新页面，查看有没有信息更新，这当然是很"low"且机械无趣的工作（当然百万粉丝的大V们可能并不同意此观点），而且更糟糕的是如果你要保持消息能及时更新，会增加轮询的频率，这无疑会加重内容提供方服务器的负担。

可能有人会说好像并不是所有信息都是轮询方式查询啊，比如电子邮件客户端。而且CSDN等网站的消息也可以通过浏览器推送了嘛。

但是这其实有很大一部分都是“假像”，其本质都是用程序自己轮询代替人肉，一旦发现信息更新就会弹出一个像是推送的“消息通知”，但其本质上都是轮询。

扯一个离题远的例子，因为众所周知的原因，国内的安卓手机是没有GSM服务的，而这带来的一个额外问题是安卓上的消息推送服务业一并没了，那国内的安卓机怎么获取消息推送呢？或许你已经得知答案了——轮询，即APP必须一直挂在后台进行轮询服务器，询问服务器有没有新消息。而这也就是为什么国内的安卓机更费电...

不过几年前小米等厂商联合搞了个国内的消息推送平台，其目的是解决这个痛点，我没有关注此事，所以就不继续扯了。

说回来，比起轮询这种简单粗暴，但效率低下的方式，观察者模式无疑更优雅，它会在你关注的信息更新后“自动地”通知你，而你所要做的只不过是提前“订阅”你所关注的消息而已。

当然了，优雅的解决方案意味着你要付出更多的学习成本，这也就是此文的意义所在。

天气APP

这里沿用《Head First 设计模式》中的例子，使用一个关于天气信息显示的例子来说明观察者模式，只不过为了能更贴合当下的时代，使用天气APP来代替书中的老旧的天气显示面板之类的概念。

假设现在我们有这样的一个需求：我们有一个信息提供方提供的天气信息获取SDK，通过这个SDK我们可以抓取天气信息，而我们的工作是要使用这些数据创建几个客户定制的天气APP。

假设天气SDK的内容是下面这样的：

# weather_sdk.py
from enum import Enum, unique


@unique
class AirQuality(Enum):
    EXCELLENT = 1
    GOODE = 2
    NORMAL = 3
    LIGHTLY_POLLUTED = 4
    HEAVEY_POLLUTED = 5


def show_air_quality_text(airQuality: AirQuality) -> str:
    text: str = ''
    if airQuality == AirQuality.EXCELLENT:
        text = '优秀'
    elif airQuality == AirQuality.GOODE:
        text = '良好'
    elif airQuality == AirQuality.NORMAL:
        text = '一般'
    elif airQuality == AirQuality.LIGHTLY_POLLUTED:
        text = '轻度污染'
    elif airQuality == AirQuality.HEAVEY_POLLUTED:
        text = '重度污染'
    else:
        text = '未定义的空气质量'
    return text


class WeatherSDK:
    def __init__(self) -> None:
        self.temperature: float = 0
        self.humidity: float = 0
        self.airQuality: AirQuality = AirQuality.NORMAL

    def weatherChanged():
        '''天气信息如果改变，此方法会被调用'''
        pass

我们创建的APP1是这样的：

#weather_app1.py
from .weather_sdk import AirQuality, show_air_quality_text


class WeatherApp1:
    def __init__(self) -> None:
        self.temperature: float = 0
        self.humidity: float = 0
        self.airQuality: AirQuality = AirQuality.NORMAL

    def update(self, temperature: float, humidity: float, airQuality: AirQuality):
        '''更新天气信息'''
        self.temperature = temperature
        self.humidity = humidity
        self.airQuality = airQuality
        self.display()

    def display(self) -> None:
        '''显示天气信息'''
        print("="*10)
        print("天气APP1")
        print("现在的气温：{:.2f}摄氏度".format(self.temperature))
        print("现在的湿度{:.2f}".format(self.humidity))
        print("现在的空气质量{}".format(show_air_quality_text(self.airQuality)))
        print("="*10)

SDK在天气数据改变后会调用weatherChanged方法，所以为了在APP1中实时更新数据，最简单的方式是这样：

    def __weatherChanged(self):
        '''天气信息如果改变，此方法会被调用'''
        app1 = WeatherApp1()
        app1.update(self.temperature, self.humidity, self.airQuality)

为了进行测试，给SDK添加了一个changeWeatherInfo方法用于修改天气数据：

    def changeWeatherInfo(self, temperature: float, humidity: float, airQuality: AirQuality) -> None:
        self.temperature = temperature
        self.humidity = humidity
        self.airQuality = airQuality
        self.__weatherChanged()

进行测试：

from src.weather_sdk import WeatherSDK, AirQuality
sdk = WeatherSDK()
sdk.changeWeatherInfo(1, 2, AirQuality.HEAVEY_POLLUTED)
# ==========
# 天气APP1
# 现在的气温：1.00摄氏度
# 现在的湿度2.00
# 现在的空气质量重度污染
# ==========

这样做当然可以实现功能，但是在WeatherSDK中我们用硬编码的方式直接使用了APP1，这样做显然是一种高度耦合的方式，而且如果要添加其他样式的APP，我们也不得不在WeatherSDK中编写更多类似的代码，这样只会给后期的维护和扩展带来无比糟糕的体验。

而使用观察者模式就可以解决这些问题。事实上，设计模式就是对解决方案中的模块进行进一步抽象，从而解耦后的更优的解决方案。

解耦往往是一种更深层次的抽象的过程。

解耦

要想解耦，需要先分析目前高耦合度的方案中哪些东西是可变的，哪些东西是共通的，可以进行深层次抽象的概念。

为了分析和说明上面的问题，我们需要实现其他的几个天气APP：

#weather_app2.py

class WeatherApp2:
    def __init__(self) -> None:
        self.humidity: float = 0
    
    def update(self, humidity: float):
        '''更新天气信息'''
        self.humidity = humidity
        self.display()

    def display(self) -> None:
        '''显示天气信息'''
        print("="*10)
        print("天气APP2")
        print("现在的湿度{:.2f}".format(self.humidity))
        rainHint: str = ""
        if self.humidity > 10:
            rainHint = "可能要下雨了，请带上雨具"
        else:
            rainHint = "不会下雨，放心出去浪吧"
        print("下雨提示：{}".format(rainHint))
        print("="*10)

# weather_app3.py
from .air_quality import AirQuality,show_air_quality_text
class WeatherApp3:
    def __init__(self) -> None:
        self.airQuality: AirQuality = AirQuality.NORMAL

    def update(self, airQuality: AirQuality):
        '''更新天气信息'''
        self.airQuality = airQuality
        self.display()

    def display(self) -> None:
        '''显示天气信息'''
        print("="*10)
        print("天气APP3")
        print("现在的空气质量{}".format(show_air_quality_text(self.airQuality)))
        sportsHint: str = ""
        if self.airQuality in (AirQuality.HEAVEY_POLLUTED, AirQuality.LIGHTLY_POLLUTED):
            sportsHint = "空气质量不好，建议宅家"
        else:
            sportsHint = "空气不错，出去运动一下吧"
        print("运动建议：{}".format(sportsHint))
        print("="*10)

相应的，我们同样要修改SDK中的__weatherChanged方法，以更新额外增加的两个APP的天气数据：

    def __weatherChanged(self):
        '''天气信息如果改变，此方法会被调用'''
        app1 = WeatherApp1()
        app2 = WeatherApp2()
        app3 = WeatherApp3()
        app1.update(self.temperature, self.humidity, self.airQuality)
        app2.update(self.humidity)
        app3.update(self.airQuality)

重新测试：

from src.weather_sdk import WeatherSDK, AirQuality
sdk = WeatherSDK()
sdk.changeWeatherInfo(1, 2, AirQuality.HEAVEY_POLLUTED)
# ==========
# 天气APP1
# 现在的气温：1.00摄氏度
# 现在的湿度2.00
# 现在的空气质量重度污染
# ==========
# ==========
# 天气APP2
# 现在的湿度2.00
# 下雨提示：不会下雨，放心出去浪吧
# ==========
# ==========
# 天气APP3
# 现在的空气质量重度污染
# 运动建议：空气质量不好，建议宅家
# ==========

可以看到，三个天气APP从SDK所需的数据不同，输出的样式也不同，但有一点是相同的，即它们都需要提供一个动作以在SDK天气数据改变的时候进行调用来进行数据更新，进而在更新后输出新的天气信息。

具体到这个例子就是APP都有的方法update，而我们将具有update()方法的可以定期通知数据更新的这一类对象可以抽象为一个接口Observer，即“观察者”。而具体持有数据，并需要在所持有的数据改变后更新相关的观察者的类，在当前这个例子中显然就是天气SDK，我们可以抽象为另一个接口Subject，即“主题”。

而观察者和主题正是构成了经典观察者设计模式中的两大核心概念。

我们可以通过UML来说明。

UML

这个UML很简单，核心概念是Subject和Observer。Display的抽象其实并不是很必要。

Subject与Observer是一对多的关系，一个主题对应多个观察者，主题的数据更新后会通知关联的所有观察者。

具体主题通过registeObserver注册观察者，相当于观察者订阅了这个主题。deleteObserver用于将观察者从订阅的主题上删除，notifyObservers用于通知所有的订阅了当前主题的观察者数据已更新。

我们这个例子中的具体类都将通过实现这两个接口的方式：

WeatherSDK实现Subject接口，其具体通过一个List之类的数据结构持有多个Observer对象，这在Java中可能是LinkedList或ArrayList，在Python中可能就是list，但这些并不重要，我们只需要明确其通过某种容器持有多个Observer对象即可。
WeatherAPP1、WeatherAPP2、WeatherAPP3实现了Observer接口，提供一个方法update()给Subject，用于被提醒更新数据。如果需要定制更多的WeatherAPP也很容易，只要继续实现Observer接口即可。

“推”和“拉”

明确了UML图以后我们就可以进行具体实现了，这里我使用EA直接生成代码框架，然后进行修改。这里只展示APP1以及WeatherSDK的代码，更多代码请前往本系列文章的Github仓库自行查看。

#######################################################
#
# WeatherAPP1.py
# Python implementation of the Class WeatherAPP1
# Generated by Enterprise Architect
# Created on:      13-6��-2021 18:19:09
# Original author: 70748
#
#######################################################
from .WeatherSDK import WeatherSDK
from .Subject import Subject
from .Observer import Observer
from .Display import Display
from .air_quality import AirQuality,show_air_quality_text


class WeatherAPP1(Observer, Display):
    def __init__(self) -> None:
        super().__init__()
        self.temperature: float = 0
        self.humidity: float = 0
        self.airQuality: AirQuality = AirQuality.NORMAL

    def display(self):
        '''显示天气信息'''
        print("="*10)
        print("天气APP1")
        print("现在的气温：{:.2f}摄氏度".format(self.temperature))
        print("现在的湿度{:.2f}".format(self.humidity))
        print("现在的空气质量{}".format(show_air_quality_text(self.airQuality)))
        print("="*10)

    def update(self, subject: Subject):
        if isinstance(subject, WeatherSDK):
            sdk: WeatherSDK = subject
            self.temperature = sdk.temperature
            self.humidity = sdk.humidity
            self.airQuality = sdk.airQuality
            self.display()

#######################################################
#
# WeatherSDK.py
# Python implementation of the Class WeatherSDK
# Generated by Enterprise Architect
# Created on:      13-6��-2021 18:19:09
# Original author: 70748
#
#######################################################
from typing import List
from .Observer import Observer
from .Subject import Subject
from .air_quality import AirQuality


class WeatherSDK(Subject):
    def __init__(self) -> None:
        super().__init__()
        self.observers: List[Observer] = []
        self.temperature: float = 0
        self.humidity: float = 0
        self.airQuality: AirQuality = AirQuality.NORMAL

    def changeWeatherInfo(self, temperature: float, humidity: float, airQuality: AirQuality) -> None:
        self.temperature = temperature
        self.humidity = humidity
        self.airQuality = airQuality
        self.notifyObservers()

    def deleteObserver(self, observer: Observer):
        self.observers.remove(observer)

    def notifyObservers(self):
        for observer in self.observers:
            observer.update(self)

    def registeObserver(self, observer: Observer):
        self.observers.append(observer)

# test.py
from src.WeatherAPP1 import WeatherAPP1
from src.WeatherAPP2 import WeatherAPP2
from src.WeatherAPP3 import WeatherAPP3
from src.WeatherSDK import WeatherSDK
from src.air_quality import AirQuality
sdk = WeatherSDK()
app1 = WeatherAPP1()
app2 = WeatherAPP2()
app3 = WeatherAPP3()
sdk.registeObserver(app1)
sdk.registeObserver(app2)
sdk.registeObserver(app3)
sdk.changeWeatherInfo(1, 2, AirQuality.LIGHTLY_POLLUTED)
# ==========
# 天气APP1
# 现在的气温：1.00摄氏度
# 现在的湿度2.00
# 现在的空气质量轻度污染
# ==========
# ==========
# 天气APP2
# 现在的湿度2.00
# 下雨提示：不会下雨，放心出去浪吧
# ==========
# ==========
# 天气APP3
# 现在的空气质量轻度污染
# 运动建议：空气质量不好，建议宅家
# ==========

需要注意的是，这里我们通过Observer子类的def update(self, subject: Subject):方法，在SDK的数据改变后，直接将其实例引用传入，从而让Observer的具体子类获取到需要的天气数据。

update方法接收的参数类型是Subject，而之后用isinstance进一步判断是否为其子类WeatherSDK的类型，然后再使用的方式是有意为之，这样做可以让方法的使用限制更宽泛，而不是仅仅先定义于WeatherSDK类型。

这种方式我们可以称之为“推”。

除了这种方式以外，我们可以让Observer对象直接持有Subject的引用，这样就可以在调用update的时候无需传递数据，直接让Observer从持有的句柄自行拉取数据即可。

我们可以称呼这种方式为“拉”。

具体的修改方式很简单，修改WeatherAPP!的初始化方法，接收一个Subject的句柄并持有：

class WeatherAPP1(Observer, Display):
    def __init__(self, subject:Subject) -> None:
        super().__init__()
        self.temperature: float = 0
        self.humidity: float = 0
        self.airQuality: AirQuality = AirQuality.NORMAL
        self.subject: Subject = subject

在update方法中直接使用持有的Subject实例：

    def update(self):
        if isinstance(self.subject, WeatherSDK):
            sdk: WeatherSDK = self.subject
            self.temperature = sdk.temperature
            self.humidity = sdk.humidity
            self.airQuality = sdk.airQuality
            self.display()

在WeatherSDK中调用update时候也无需在传递数据：

    def notifyObservers(self):
        for observer in self.observers:
            observer.update()

完整代码见Github仓库的pattern2/weather_v3

这里关于观察者模式的全部内容其实已经介绍完毕了，下面探讨符合Python风格的观察者模式应该如何编写。

Python式

如同在Python学习笔记23：Python设计模式中介绍的那样，Python并不需要形式上的接口，即无需强行定义一个确实存在的接口，让子类继承。Python的方式往往是更宽泛的“协议”。即通过文档人为约定实现了哪些方法以作为一种协议，实现相应的方法即可看做是实现了某种协议。

具体到这里的例子，我们可以无需实现Subject和Observer接口，只要讲它们看做是两种协议即可。

此外，Subject持有多个Observer的目的无非是为了通知数据变更，从这个角度上说，完全可以只持有update方法，作为一种回调函数。因为Python中的函数是一类对象，所以这样做完全是可行的。

具体代码就不在这里展示了，完整代码见Github仓库的pattern2/weather_v4。

如果要Subject直接持有update方法，需要修改大量代码，所以在weather_v4中我并没有这么做。

这样做无疑少了很多样板代码，但是同样的，也不是没有代价。代码中原来很明确的类型提示，比如subject:Subject，已经变成了subject:"Subject"，这样会导致IDE无法对相应的变量提供智能联想功能。所以在Python的类型提示功能已经相当完善的今天，编写一些不那么Python风格的基类代码或许也并不是那么一无是处。

关于Python的类型提示功能，可以阅读PEP 484 -- Type Hints。

好了，关于观察者模式的内容全部介绍完毕，谢谢阅读。

本系列的所有示例代码及相关文档都保存在Github项目 design-pattern-with-python。

本作品采用知识共享署名 4.0 国际许可协议进行许可

设计模式 with Python2：观察者模式

概念

天气APP

解耦

UML

“推”和“拉”

Python式

文章评论