Python 事件管理器 (EventManager) 实现:使用多线程监听事件
以下是一个简单的 EventManager 的实现,其中包括 AddEventListener 和 AddEvent 方法,并实现了预设事件 'ticking'。
import threading
class EventManager:
def __init__(self):
self.event_listeners = {}
def add_event_listener(self, event_name, callback):
if event_name not in self.event_listeners:
self.event_listeners[event_name] = []
self.event_listeners[event_name].append(callback)
def add_event(self, event_name, data):
if event_name in self.event_listeners:
callbacks = self.event_listeners[event_name]
for callback in callbacks:
threading.Thread(target=callback, args=(data,)).start()
def update_list(data):
print('Updating list with data:', data)
def main():
event_manager = EventManager()
event_manager.add_event_listener('ticking', update_list)
for i in range(100):
event_manager.add_event('ticking', [i])
if __name__ == '__main__':
main()
在上述示例中,我们首先创建了一个 EventManager 类,它有一个 event_listeners 字典来存储事件和回调函数之间的映射关系。在 add_event_listener 方法中,我们通过将回调函数添加到相应事件的列表中,来创建事件和回调函数之间的关联。在 add_event 方法中,我们检查事件是否有对应的回调函数,并对每个回调函数创建一个新的线程来执行。
在 main 函数中,我们首先创建了一个 EventManager 实例,并使用 add_event_listener 方法将 'ticking' 事件与 update_list 回调函数关联起来。然后,我们使用一个简单的循环来模拟触发 'ticking' 事件,每次触发时传递一个列表作为参数。
请注意,由于在 add_event 方法中使用了多线程,所以回调函数 update_list 可能会被并发执行。如果在回调函数中修改共享的资源,请确保使用适当的线程同步机制(例如锁)来避免竞争条件。在这个简单的例子中,我们只是打印了传递的数据,因此不需要考虑线程锁。
原文地址: https://www.cveoy.top/t/topic/qCyA 著作权归作者所有。请勿转载和采集!