事半功倍？别再说你不懂设计模式里的观察者模式！

2022-11-19 00:30:23

观察者模式与发布订阅模式：掌握编程界的多人协作

在软件开发中，你是否遇到过这样的情况：当一个对象的状态发生改变时，你需要其他对象做出相应的反应？传统的做法是让这些对象紧密耦合在一起，当一个对象发生改变时，其他对象就会直接收到通知。然而，这种方法存在一个重大的缺陷：随着系统的复杂性增加，对象之间的耦合度也会变得很高，维护起来极其困难。

观察者模式 就是为了解决这个问题而诞生的。它是一种设计模式，允许对象之间松散耦合。当一个对象发生改变时，它只需要通知它的观察者，观察者再做出相应的反应。这样一来，对象之间的耦合度就大大降低，系统的可维护性也得到了提升。

如何理解观察者模式

想象一下你在一家餐馆里，作为顾客的你是一位观察者，而服务员是你要观察的对象。当你想要点菜时，你不会直接走到厨房里，而是告诉服务员你的需求。服务员再把你的需求传达给厨房，厨房收到你的订单后就会开始准备菜品。这样一来，你作为顾客只需要关注服务员，而不用关心厨房的具体运作细节。这就是观察者模式的原理。

代码示例：观察者模式

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):
        for observer in self._observers:
            observer.update()


class ConcreteSubject(Subject):
    def __init__(self, state):
        super().__init__()
        self._state = state

    @property
    def state(self):
        return self._state

    @state.setter
    def state(self, new_state):
        self._state = new_state
        self.notify()


class Observer:
    def update(self):
        pass


class ConcreteObserverA(Observer):
    def update(self):
        print("ObserverA notified, state is:", subject.state)


class ConcreteObserverB(Observer):
    def update(self):
        print("ObserverB notified, state is:", subject.state)


if __name__ == "__main__":
    subject = ConcreteSubject(0)
    observer1 = ConcreteObserverA()
    observer2 = ConcreteObserverB()
    subject.attach(observer1)
    subject.attach(observer2)

    subject.state = 1
    subject.state = 2

发布订阅模式

发布订阅模式 与观察者模式非常相似，但也有其独特之处。在观察者模式中，观察者必须知道被观察者的身份，而在发布订阅模式中，订阅者并不知道发布者的身份。发布者只需将消息发布到消息总线上，订阅者只需要订阅消息总线上的消息，就可以接收消息。

如何理解发布订阅模式

发布订阅模式就像一个广播电台。电台（发布者）会不断播放音乐（消息），而听众（订阅者）可以随时调到这个电台（订阅消息总线）来收听音乐（接收消息）。电台并不知道每个听众的身份，而听众也只需要知道电台的频率，并不需要知道电台的具体位置和运作方式。

代码示例：发布订阅模式

import asyncio

class Publisher:
    def __init__(self):
        self.subscribers = []

    def subscribe(self, subscriber):
        self.subscribers.append(subscriber)

    def publish(self, message):
        for subscriber in self.subscribers:
            subscriber.receive(message)


class Subscriber:
    def __init__(self, name):
        self.name = name

    async def receive(self, message):
        print(f"Subscriber {self.name} received message: {message}")


async def main():
    publisher = Publisher()
    subscriber1 = Subscriber("Sub1")
    subscriber2 = Subscriber("Sub2")

    publisher.subscribe(subscriber1)
    publisher.subscribe(subscriber2)

    publisher.publish("Hello, world!")

asyncio.run(main())