Python中的观察者模式：从基础到实战

news2024/10/6 14:38:04

引言

在软件开发中，设计模式是解决特定问题的通用解决方案。观察者模式（Observer Pattern）是其中一种非常有用的模式，它定义了对象之间的一对多依赖关系，当一个对象状态发生变化时，所有依赖于它的对象都会得到通知并自动更新。这种模式在事件驱动的程序设计中尤为重要，例如GUI应用程序、网络通信、游戏开发等领域。

基础语法介绍

观察者模式包含两个主要角色：观察者（Observer）和被观察者（Subject）。被观察者维护一个观察者列表，并提供添加、删除和通知观察者的方法。观察者则需要实现一个更新接口，以便在被观察者状态变化时接收通知。

在Python中，观察者模式可以通过多种方式实现，但最常用的是使用回调函数。下面是一个基础的观察者模式实现：

class Subject:
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    def notify(self, message):
        for observer in self._observers:
            observer.update(message)

class Observer:
    def update(self, message):
        raise NotImplementedError("Subclasses should implement this!")

class ConcreteObserver(Observer):
    def update(self, message):
        print(f"Observer received message: {message}")

# 使用
subject = Subject()
observer = ConcreteObserver()
subject.attach(observer)
subject.notify("Hello, Observer Pattern!")

基础实例

问题描述

假设我们有一个新闻发布系统，当有新新闻发布时，所有订阅者都应该收到通知。

代码示例

class NewsPublisher(Subject):
    def __init__(self):
        super().__init__()
        self._news = ""

    def set_news(self, news):
        self._news = news
        self.notify(self._news)

class NewsSubscriber(Observer):
    def update(self, news):
        print(f"News received: {news}")

# 使用
publisher = NewsPublisher()
subscriber = NewsSubscriber()
publisher.attach(subscriber)
publisher.set_news("Breaking News: Python 4.0 Released!")

进阶实例

问题描述

在更复杂的系统中，我们可能需要根据不同的条件通知不同的观察者。例如，一个股票交易系统，只有当特定股票的价格变动超过一定阈值时，才通知投资者。

高级代码实例

class StockPublisher(Subject):
    def __init__(self):
        super().__init__()
        self._stocks = {}

    def update_stock(self, stock, price):
        if self._stocks.get(stock, 0) != price:
            self._stocks[stock] = price
            self.notify(stock, price)

class StockSubscriber(Observer):
    def __init__(self, threshold):
        self.threshold = threshold

    def update(self, stock, price):
        if abs(price - self.threshold) > 0.05:
            print(f"Stock {stock} price changed to {price}, notifying!")

# 使用
publisher = StockPublisher()
subscriber = StockSubscriber(150)
publisher.attach(subscriber)
publisher.update_stock("AAPL", 155)

实战案例

问题描述

在开发一个电子商务网站时，我们需要实现一个功能，当用户将商品添加到购物车时，购物车页面应该实时更新。

解决方案

我们可以使用观察者模式来实现这个功能。购物车是被观察者，而购物车页面是观察者。

代码实现

class ShoppingCart(Subject):
    def __init__(self):
        super().__init__()
        self._items = []

    def add_item(self, item):
        self._items.append(item)
        self.notify(self._items)

class CartPage(Observer):
    def update(self, items):
        print(f"Cart updated with items: {items}")

# 使用
cart = ShoppingCart()
page = CartPage()
cart.attach(page)
cart.add_item("Laptop")
cart.add_item("Mouse")