请用 Python 写一个 EventManager该EventManager将启动独立thread并监听各种事件请添加如AddEventListener与AddEvent的方法。对于你将回答的一个实现请添加预设事件ticking该ticking事件将被使用示例中的一个for i in range100的循环中不断触发并传递一个列表作为参数。使用示例中触发该事件将调用回调函数update_list
以下是一个简单的EventManger的实现,其中包括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可能会被并发执行。如果在回调函数中修改共享的资源,请确保使用适当的线程同步机制(例如锁)来避免竞争条件。在这个简单的例子中,我们只是打印了传递的数据,因此不需要考虑线程锁
原文地址: http://www.cveoy.top/t/topic/iWwM 著作权归作者所有。请勿转载和采集!