### Python 中间件实现订阅发布模式 在 Python 中构建中间件以支持订阅/发布（Pub/Sub）模式可以通过多种方式完成。这里展示了一种基于 `threading` 和自定义类的方法，这种方法可以用于创建轻量级的消息传递机制。 #### 定义 Publisher 类 Publisher 负责向所有注册过的 Subscriber 发送消息： ```python import threading class Publisher: def __init__(self, events): self.subscribers = {event: [] for event in events} def subscribe(self, event, subscriber): self.subscribers[event].append(subscriber) def unsubscribe(self, event, subscriber): self.subscribers[event].remove(subscriber) def publish(self, event, message): if event not in self.subscribers: return threads = [ threading.Thread(target=subscriber.update, args=(message,)) for subscriber in self.subscribers[event] ] for thread in threads: thread.start() for thread in threads: thread.join() ``` 此部分代码展示了如何设置一个简单的 Publish-Subscribe 架构中的 Publisher 组件[^1]。 #### 创建 Subscriber 基础类 Subscriber 需要继承此类，并重写 update 方法来处理接收到的信息： ```python from abc import ABCMeta, abstractmethod class Subscriber(metaclass=ABCMeta): @abstractmethod def update(self, message): pass ``` 这段代码提供了 Subscriber 的基本结构，任何想要成为订阅者的实体都需要扩展这个基类并提供具体的逻辑去响应来自 publisher 的通知[^3]。 #### 使用案例 下面是一个完整的例子，说明了上述组件是如何协同工作的： ```python # 自定义的具体订阅者 class EmailAlerts(Subscriber): def update(self, message): print(f"Email alert received: {message}") class SMSAlerts(Subscriber): def update(self, message): print(f"SMS alert sent with content: {message}") if __name__ == "__main__": pub = Publisher(events=["alerts", "notifications"]) email_alerts = EmailAlerts() sms_alerts = SMSAlerts() # 注册监听器 pub.subscribe("alerts", email_alerts) pub.subscribe("alerts", sms_alerts) # 发布一条警报信息 pub.publish("alerts", "System is down!") ``` 在这个实例中，每当 Publisher 对象调用其 `publish()` 函数时，所有的 Subscribers 将会并发地执行它们各自的 `update()` 方法来处理传入的数据[^2]。