python事件触发机制_python模拟事件触发机制详解

Love The Way You Lie 2022-10-31 13:53 312阅读 0赞

本文实例为大家分享了python模拟事件触发机制的具体代码，供大家参考，具体内容如下

EventManager.py

\# -\*- encoding: UTF-8 -\*-

\# 系统模块

from queue import Queue, Empty

from threading import \*

class EventManager:

def \_\_init\_\_(self):

"""初始化事件管理器"""

\# 事件对象列表

self.\_\_eventQueue = Queue()

\# 事件管理器开关

self.\_\_active = False

\# 事件处理线程

self.\_\_thread = Thread(target = self.\_\_Run)

\# 这里的\_\_handlers是一个字典，用来保存对应的事件的响应函数

\# 其中每个键对应的值是一个列表，列表中保存了对该事件监听的响应函数，一对多

self.\_\_handlers = \{\} \# \{事件类型:\[处理事件的方法\]\}

def \_\_Run(self):

"""引擎运行"""

while self.\_\_active == True:

try:

\# 获取事件的阻塞时间设为1秒

event = self.\_\_eventQueue.get(block = True, timeout = 1)

self.\_\_EventProcess(event)

except Empty:

pass

def \_\_EventProcess(self, event):

"""处理事件"""

\# 检查是否存在对该事件进行监听的处理函数

if event.type\_ in self.\_\_handlers:

\# 若存在，则按顺序将事件传递给处理函数执行

for handler in self.\_\_handlers\[event.type\_\]:

handler(event)

def Start(self):

"""启动"""

\# 将事件管理器设为启动

self.\_\_active = True

\# 启动事件处理线程

self.\_\_thread.start()

def Stop(self):

"""停止"""

\# 将事件管理器设为停止

self.\_\_active = False

\# 等待事件处理线程退出

self.\_\_thread.join()

def AddEventListener(self, type\_, handler):

"""绑定事件和监听器处理函数"""

\# 尝试获取该事件类型对应的处理函数列表，若无则创建

try:

handlerList = self.\_\_handlers\[type\_\]

except KeyError:

handlerList = \[\]

self.\_\_handlers\[type\_\] = handlerList

\# 若要注册的处理器不在该事件的处理器列表中，则注册该事件

if handler not in handlerList:

handlerList.append(handler)

def RemoveEventListener(self, type\_, handler):

"""移除监听器的处理函数"""

\#读者自己试着实现

def SendEvent(self, event):

"""发送事件，向事件队列中存入事件"""

self.\_\_eventQueue.put(event)

"""事件对象"""

class Event:

def \_\_init\_\_(self, type\_=None):

self.type\_ = type\_ \# 事件类型

self.dict = \{\} \# 字典用于保存具体的事件数据

test.py

\# -\*- encoding: UTF-8 -\*-

from threading import \*

from EventManager import \*

import time

\#事件名称 新文章

EVENT\_ARTICAL = "Event\_Artical"

\#事件源 公众号

class PublicAccounts:

def \_\_init\_\_(self,eventManager):

self.\_\_eventManager = eventManager

def WriteNewArtical(self):

\#事件对象，写了新文章

event = Event(type\_=EVENT\_ARTICAL)

event.dict\["artical"\] = u'如何写出更优雅的代码\\n'

\#发送事件

self.\_\_eventManager.SendEvent(event)

print(u'公众号发送新文章')

\#监听器 订阅者

class Listener:

def \_\_init\_\_(self,username):

self.\_\_username = username

\#监听器的处理函数 读文章

def ReadArtical(self,event):

print(u'%s 收到新文章' % self.\_\_username)

print(u'正在阅读新文章内容：%s' % event.dict\["artical"\])

"""测试函数"""

def test():

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)

while True:

publicAcc.WriteNewArtical()

time.sleep(2)

if \_\_name\_\_ == '\_\_main\_\_':

test()

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。

python事件触发机制_python模拟事件触发机制详解

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 MySQL基于时间的事件触发机制实例

相关 JS 事件触发机制

相关回车键触发事件

相关 python事件触发机制_python模拟事件触发机制详解

相关回车触发事件

相关回车触发事件

相关 jquery事件触发

相关 jquery事件触发

相关 jQuery_触发事件

相关 javascript模拟（触发）鼠标事件