以下是一个简单的 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 可能会被并发执行。如果在回调函数中修改共享的资源,请确保使用适当的线程同步机制(例如锁)来避免竞争条件。在这个简单的例子中,我们只是打印了传递的数据,因此不需要考虑线程锁。

Python 事件管理器 (EventManager) 实现:使用多线程监听事件

原文地址: https://www.cveoy.top/t/topic/qCyA 著作权归作者所有。请勿转载和采集!

免费AI点我,无需注册和登录