下面是一个实现发布-订阅（Publisher-Subscriber）系统的 Python 模板代码，它采用模块化设计和功能扩展性，支持动态注册、取消订阅以及快速集成新功能：
from typing import Callable, Dict, List, Any
class Event:
"""
Represents an event type in the Pub-Sub system.
"""
def __init__(self, name: str):
self.name = name
def __repr__(self):
return f"Event(name={self.name})"
class Subscriber:
"""
Base class for subscribers. Can be extended for specific behaviors.
"""
def __init__(self, name: str):
self.name = name
def notify(self, event: Event, data: Any):
"""
Action performed when the subscriber receives an event notification.
Override this method to define custom behavior.
"""
print(f"{self.name} received event '{event.name}' with data: {data}")
class Publisher:
"""
Publisher class to manage events and subscribers.
"""
def __init__(self):
self._subscribers: Dict[str, List[Callable]] = {}
def register(self, event_name: str, callback: Callable):
"""
Registers a subscriber (callback function) to an event.
"""
if event_name not in self._subscribers:
self._subscribers[event_name] = []
self._subscribers[event_name].append(callback)
print(f"Registered callback for event '{event_name}'.")
def unregister(self, event_name: str, callback: Callable):
"""
Unregisters a subscriber (callback function) from an event.
"""
if event_name in self._subscribers:
try:
self._subscribers[event_name].remove(callback)
print(f"Unregistered callback for event '{event_name}'.")
if not self._subscribers[event_name]:
del self._subscribers[event_name] # Remove event if no subscribers
except ValueError:
print(f"Callback not found for event '{event_name}'.")
else:
print(f"No such event '{event_name}' to unregister from.")
def notify(self, event_name: str, data: Any):
"""
Notifies all subscribers of a particular event with provided data.
"""
if event_name in self._subscribers:
print(f"Notifying subscribers of event '{event_name}'...")
for callback in self._subscribers[event_name]:
callback(data)
else:
print(f"No subscribers to notify for event '{event_name}'.")
# Example: Testing the Pub-Sub System
if __name__ == "__main__":
# Create a Publisher
pub = Publisher()
# Create Subscribers
alice = Subscriber("Alice")
bob = Subscriber("Bob")
# Define Event Handlers (callbacks)
def alice_handler(data):
alice.notify(Event("TestEvent"), data)
def bob_handler(data):
bob.notify(Event("TestEvent"), data)
# Register Handlers
pub.register("TestEvent", alice_handler)
pub.register("TestEvent", bob_handler)
# Trigger Events
pub.notify("TestEvent", {"message": "Hello, World!"})
# Unregister one subscriber
pub.unregister("TestEvent", alice_handler)
# Trigger Events again
pub.notify("TestEvent", {"message": "Goodbye, World!"})
代码说明
-
模块化设计：
Event 类表示事件对象（可以扩展为更复杂的事件数据结构）。Subscriber 类是一个基础类，支持自定义行为（通过重写 notify 方法）。Publisher 类是核心，负责维护事件与回调的关系。
-
动态注册与取消订阅：
- 使用
register 方法动态添加回调函数。
- 使用
unregister 方法在运行时移除回调。
-
扩展性与适配器模式：
Subscriber 可以被子类化，添加特定的通知行为。- 使用回调函数机制可以适配任何类型的事件处理函数，甚至可以对接更复杂的系统。
-
性能优化与可维护性：
- 支持按需添加和移除订阅者，避免冗余通知。
- 事件键值基于动态字典存储，支持任意数量的事件类型。
如何扩展
- 添加事件优先级：引入优先级队列，根据 Subscriber 的优先级排序触发回调。
- 异步通知：使用 Python 的
asyncio 或线程池实现异步通知机制。
- 持久化事件日志：扩展
Publisher 类，将事件通知历史记录存储到文件或数据库中。
- 订阅过滤：在
notify 方法中添加数据过滤规则，让特定数据只发送到符合条件的订阅者。
这是一个简单但功能强大的发布订阅系统模板，可以根据业务需求进一步定制和扩展。
