正文

Python编程：轻松掌握观察者模式，附实战示例代码详解

/0 浏览量
1018

观察者模式（Observer Pattern）是一种设计模式，它定义了一种一对多的依赖关系，当一个对象的状态发生改变时，所有依赖于它的对象都会得到通知并自动更新。在Python中，我们可以通过多种方式实现观察者模式，包括使用装饰器、类和标准库中的功能。

观察者模式的基本概念

在观察者模式中，通常有两个角色：

  • Subject（被观察者）：它负责维护一个观察者列表，并提供注册、移除和通知观察者的接口。
  • Observer（观察者）：它订阅Subject的状态变化，并在状态变化时执行相应的操作。

Python实现观察者模式

使用装饰器实现观察者模式

装饰器是一种非常灵活的方式来实现观察者模式。以下是一个使用装饰器实现观察者模式的例子：

def observer(func):
    def wrapper(*args, **kwargs):
        func(*args, **kwargs)
    return wrapper

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

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

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

    def notify(self):
        for observer in self._observers:
            observer()

@observer
def observer_function():
    print("Observer received notification!")

subject = Subject()
subject.register(observer_function)
subject.notify()  # 输出：Observer received notification!

使用类实现观察者模式

另一种方式是直接使用类来定义Subject和Observer：

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

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

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

    def notify(self, *args, **kwargs):
        for observer in self._observers:
            observer(self, *args, **kwargs)

class Observer:
    def update(self, subject, *args, **kwargs):
        pass

class ConcreteObserver(Observer):
    def update(self, subject, *args, **kwargs):
        print(f"Observer received notification from {subject.__class__.__name__} with args: {args}, kwargs: {kwargs}")

subject = Subject()
observer1 = ConcreteObserver()
observer2 = ConcreteObserver()

subject.register(observer1)
subject.register(observer2)

subject.notify(1, 2, 3)  # 输出：Observer received notification from Subject with args: (1, 2, 3), kwargs: {}

使用标准库实现观察者模式

Python的标准库中提供了Observable类，可以直接使用：

from observable import Observable

class Subject(Observable):
    pass

subject = Subject()
observer1 = lambda *args, **kwargs: print(f"Observer received notification with args: {args}, kwargs: {kwargs}")

subject.register(observer1)
subject.notify(1, 2, 3)  # 输出：Observer received notification with args: (1, 2, 3), kwargs: {}

实战示例

以下是一个使用观察者模式的实战示例，模拟一个股票价格的变动，并通知所有订阅者：

class StockSubject(Observable):
    def __init__(self, price):
        super().__init__()
        self._price = price

    def set_price(self, new_price):
        if self._price != new_price:
            self._price = new_price
            self.notify(self, new_price)

class StockObserver(Observer):
    def update(self, subject, new_price):
        print(f"Stock price changed to {new_price}")

# 创建Subject实例
stock_subject = StockSubject(100)

# 创建Observer实例
stock_observer = StockObserver()

# 注册Observer
stock_subject.register(stock_observer)

# 更改价格，并通知Observer
stock_subject.set_price(105)

通过以上示例，我们可以看到观察者模式在Python中的实现方法。在实际项目中，观察者模式可以用于实现各种场景，如事件监听、状态更新等。掌握观察者模式，有助于我们写出更加灵活和可扩展的代码。

-- 展开阅读全文 --