Python 实现带取消注册功能的简单发布 - 订阅模式：EventEmitter 与 Listener 类详解

下面是一个简单的发布-订阅模式的实现，包含 `EventEmitter` 和 `Listener` 两个类。 ```python class Listener: def __init__(self, handler): self.handler = handler class EventEmitter: def __init__(self): self.events = {} def on(self, event_name, listener): if event_name not in self.events: self.events[event_name] = [] self.events[event_name].append(listener) def emit(self, event_name, *args, kwargs): if event_name in self.events: for listener in self.events[event_name]: listener.handler(*args, kwargs) def off(self, event_name, listener): if event_name in self.events: self.events[event_name] = [l for l in self.events[event_name] if l != listener] # 示例使用 def handle_event_1(*args, kwargs): print(f"Event 1 triggered with args: {args} and kwargs: {kwargs}") def handle_event_2(*args, kwargs): print(f"Event 2 triggered with args: {args} and kwargs: {kwargs}") # 创建 EventEmitter 实例 emitter = EventEmitter() # 创建 Listener 实例 listener1 = Listener(handle_event_1) listener2 = Listener(handle_event_2) listener3 = Listener(handle_event_1) # 另一个监听 event_1 的监听器 # 注册监听器 emitter.on("event_1", listener1) emitter.on("event_1", listener3) emitter.on("event_2", listener2) # 触发事件 emitter.emit("event_1", 1, 2, 3, a="foo", b="bar") emitter.emit("event_2", "hello", world="world") # 取消注册特定的监听器 emitter.off("event_1", listener1) # 再次触发事件，查看监听器是否已被取消注册 emitter.emit("event_1", 4, 5, 6) ``` 代码解释 1. Listener 类： - `__init__` 方法接收一个处理器函数 `handler` 并将其存储在实例变量中。 2. EventEmitter 类： - `__init__` 方法初始化一个字典 `events` 用于存储事件和对应的监听器列表。 - `on` 方法用于注册监听器。如果事件名称不存在于 `events` 字典中，则创建一个空列表。然后将监听器添加到对应事件的列表中。 - `emit` 方法用于触发事件。如果事件名称存在于 `events` 字典中，则遍历该事件的监听器列表并调用每个监听器的 `handler` 方法。 - `off` 方法用于取消注册特定的监听器。通过遍历监听器列表并移除指定的监听器来实现。 示例使用 - 创建 `EventEmitter` 实例。 - 创建多个 `Listener` 实例，每个实例都包含一个事件处理器函数。 - 使用 `on` 方法注册监听器。 - 使用 `emit` 方法触发事件，并传递参数。 - 使用 `off` 方法取消注册特定的监听器。 - 再次触发事件，验证监听器是否已被正确取消注册。 这样，一个简单的发布-订阅模式就实现了，可以注册多个监听同一事件的监听器，并在事件被触发时执行所有注册的监听器。也支持取消注册特定的监听器。 ######[AI写代码神器 | 809点数解答 | 2024-11-09 22:17:06]