20、设计模式：观察者模式：发布-订阅模型，事件驱动编程，实战案例：气象站数据更新

观察者模式，说白了就是「你盯着我，我变了就告诉你」。

我刚开始接触这个模式时，总觉得它跟回调函数差不多。后来在做一个实时监控系统时，才真正体会到它的威力——当时系统里有十几个模块都要监听同一个数据源的变化，如果不用观察者模式，代码会乱成一锅粥。

什么是观察者模式？

观察者模式定义了一种一对多的依赖关系。当一个对象的状态发生变化时，所有依赖它的对象都会得到通知并自动更新。

它有两个核心角色：

  • Subject（主题）：被观察的对象，负责维护观察者列表，并通知它们
  • Observer（观察者）：订阅主题的对象，当主题变化时执行自己的更新逻辑

核心思想：主题不关心观察者是谁，观察者也不关心主题内部怎么运作。它们之间只通过一个接口通信——这就是「松耦合」的精髓。

发布-订阅模型 vs 传统观察者模式

你可能会问：发布-订阅和观察者模式是不是一回事？

嗯，这里要注意。它们很像，但有一个关键区别：

特性 传统观察者模式 发布-订阅模式
通信方式 直接通知 通过消息代理
耦合度 主题知道观察者 完全解耦
适用场景 简单通知 复杂事件系统

我个人习惯在中小型项目里直接用观察者模式，简单直接。但如果是微服务架构或者跨模块通信，发布-订阅模型会更合适。

事件驱动编程

观察者模式是事件驱动编程的基石。你想想看，GUI 里的按钮点击、网络请求的回调、消息队列的消费——本质上都是观察者模式的应用。

事件驱动编程的核心流程：

  1. 注册事件处理器（观察者）
  2. 等待事件发生
  3. 事件触发时，自动调用所有注册的处理器

避坑指南：我曾经在一个项目中，事件处理器里又触发了新事件，结果形成了死循环。后来我加了一个「事件处理中」的标志位，才解决了这个问题。所以，事件处理器里尽量不要触发同类型事件。

实战案例：气象站数据更新

我们来做一个气象站的例子。气象站会不断采集温度、湿度、气压数据。多个显示面板需要实时更新这些数据。

先定义观察者接口：

from abc import ABC, abstractmethod

class Observer(ABC):
    @abstractmethod
    def update(self, temperature, humidity, pressure):
        pass

再定义主题接口：

class Subject(ABC):
    @abstractmethod
    def register_observer(self, observer):
        pass
    
    @abstractmethod
    def remove_observer(self, observer):
        pass
    
    @abstractmethod
    def notify_observers(self):
        pass

实现具体的气象站主题：

class WeatherData(Subject):
    def __init__(self):
        self.observers = []
        self.temperature = 0
        self.humidity = 0
        self.pressure = 0
    
    def register_observer(self, observer):
        self.observers.append(observer)
    
    def remove_observer(self, observer):
        self.observers.remove(observer)
    
    def notify_observers(self):
        for observer in self.observers:
            observer.update(self.temperature, self.humidity, self.pressure)
    
    def set_measurements(self, temperature, humidity, pressure):
        self.temperature = temperature
        self.humidity = humidity
        self.pressure = pressure
        self.notify_observers()

实现两个显示面板：

class CurrentConditionsDisplay(Observer):
    def update(self, temperature, humidity, pressure):
        print(f"当前温度：{temperature}°C")
        print(f"当前湿度：{humidity}%")
        print(f"当前气压：{pressure}hPa")

class StatisticsDisplay(Observer):
    def __init__(self):
        self.temps = []
    
    def update(self, temperature, humidity, pressure):
        self.temps.append(temperature)
        avg = sum(self.temps) / len(self.temps)
        print(f"平均温度：{avg:.1f}°C")

使用示例：

weather_data = WeatherData()
current_display = CurrentConditionsDisplay()
stats_display = StatisticsDisplay()

weather_data.register_observer(current_display)
weather_data.register_observer(stats_display)

weather_data.set_measurements(25, 60, 1013)
weather_data.set_measurements(26, 58, 1012)

注意：观察者模式虽然好用，但不要滥用。如果观察者太多，通知所有观察者可能会成为性能瓶颈。我建议观察者数量超过 50 个时，考虑使用异步通知或者批量处理。

观察者模式的核心逻辑图

WeatherData（主题） CurrentConditionsDisplay StatisticsDisplay ForecastDisplay register register register notify setMeasurements() 触发更新 观察者列表：[display1, display2, display3] 遍历列表，逐个调用 update()

什么时候用观察者模式？

我总结了几个典型场景：

  • 一个对象的变化需要同时更新多个其他对象
  • 你不想让主题和观察者之间产生强耦合
  • 观察者的数量可能在运行时动态变化

我的经验：观察者模式特别适合做「事件总线」。我曾经在一个游戏引擎里用观察者模式实现了成就系统——玩家做某个动作时，不需要知道哪些成就会被触发，成就系统自己会监听事件。这样加新成就时，完全不用改原有代码。

观察者模式看起来简单，但它是很多高级架构的基础。MVC 里的 Model 通知 View、React 里的状态管理、消息队列的发布订阅——底层逻辑都是观察者模式。

说白了，这个模式就干了一件事：把「谁关心这个变化」的逻辑从业务代码里抽离出来。代码更干净，扩展更方便。

公众号：蓝海资料掘金营，微信 deep3321

← 上一章 📖 返回目录 下一章 →