python模拟事件触发机制详解
本文实例为大家分享了python模拟事件触发机制的具体代码,供大家参考,具体内容如下
EventManager.py
#-*-encoding:UTF-8-*- #系统模块 fromqueueimportQueue,Empty fromthreadingimport* classEventManager: def__init__(self): """初始化事件管理器""" #事件对象列表 self.__eventQueue=Queue() #事件管理器开关 self.__active=False #事件处理线程 self.__thread=Thread(target=self.__Run) #这里的__handlers是一个字典,用来保存对应的事件的响应函数 #其中每个键对应的值是一个列表,列表中保存了对该事件监听的响应函数,一对多 self.__handlers={}#{事件类型:[处理事件的方法]} def__Run(self): """引擎运行""" whileself.__active==True: try: #获取事件的阻塞时间设为1秒 event=self.__eventQueue.get(block=True,timeout=1) self.__EventProcess(event) exceptEmpty: pass def__EventProcess(self,event): """处理事件""" #检查是否存在对该事件进行监听的处理函数 ifevent.type_inself.__handlers: #若存在,则按顺序将事件传递给处理函数执行 forhandlerinself.__handlers[event.type_]: handler(event) defStart(self): """启动""" #将事件管理器设为启动 self.__active=True #启动事件处理线程 self.__thread.start() defStop(self): """停止""" #将事件管理器设为停止 self.__active=False #等待事件处理线程退出 self.__thread.join() defAddEventListener(self,type_,handler): """绑定事件和监听器处理函数""" #尝试获取该事件类型对应的处理函数列表,若无则创建 try: handlerList=self.__handlers[type_] exceptKeyError: handlerList=[] self.__handlers[type_]=handlerList #若要注册的处理器不在该事件的处理器列表中,则注册该事件 ifhandlernotinhandlerList: handlerList.append(handler) defRemoveEventListener(self,type_,handler): """移除监听器的处理函数""" #读者自己试着实现 defSendEvent(self,event): """发送事件,向事件队列中存入事件""" self.__eventQueue.put(event) """事件对象""" classEvent: def__init__(self,type_=None): self.type_=type_#事件类型 self.dict={}#字典用于保存具体的事件数据
test.py
#-*-encoding:UTF-8-*- fromthreadingimport* fromEventManagerimport* importtime #事件名称新文章 EVENT_ARTICAL="Event_Artical" #事件源公众号 classPublicAccounts: def__init__(self,eventManager): self.__eventManager=eventManager defWriteNewArtical(self): #事件对象,写了新文章 event=Event(type_=EVENT_ARTICAL) event.dict["artical"]=u'如何写出更优雅的代码\n' #发送事件 self.__eventManager.SendEvent(event) print(u'公众号发送新文章') #监听器订阅者 classListener: def__init__(self,username): self.__username=username #监听器的处理函数读文章 defReadArtical(self,event): print(u'%s收到新文章'%self.__username) print(u'正在阅读新文章内容:%s'%event.dict["artical"]) """测试函数""" deftest(): listner1=Listener("thinkroom")#订阅者1 listner2=Listener("steve")#订阅者2 eventManager=EventManager() #绑定事件和监听器响应函数(新文章) eventManager.AddEventListener(EVENT_ARTICAL,listner1.ReadArtical) eventManager.AddEventListener(EVENT_ARTICAL,listner2.ReadArtical) eventManager.Start() publicAcc=PublicAccounts(eventManager) whileTrue: publicAcc.WriteNewArtical() time.sleep(2) if__name__=='__main__': test()
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持毛票票。