python 模拟事件触发机制

EventManager.py

 1 # -*- encoding: UTF-8 -*-
 2
 3 # 系统模块
 4 from queue import Queue, Empty
 5 from threading import *
 6
 7
 8 class EventManager:
 9     def __init__(self):
10         """初始化事件管理器"""
11         # 事件对象列表
12         self.__eventQueue = Queue()
13         # 事件管理器开关
14         self.__active = False
15         # 事件处理线程
16         self.__thread = Thread(target = self.__Run)
17
18         # 这里的__handlers是一个字典，用来保存对应的事件的响应函数
19         # 其中每个键对应的值是一个列表，列表中保存了对该事件监听的响应函数，一对多
20         self.__handlers = {}   # {事件类型:[处理事件的方法]}
21
22     def __Run(self):
23         """引擎运行"""
24         while self.__active == True:
25             try:
26                 # 获取事件的阻塞时间设为1秒
27                 event = self.__eventQueue.get(block = True, timeout = 1)
28                 self.__EventProcess(event)
29             except Empty:
30                 pass
31
32     def __EventProcess(self, event):
33         """处理事件"""
34         # 检查是否存在对该事件进行监听的处理函数
35         if event.type_ in self.__handlers:
36             # 若存在，则按顺序将事件传递给处理函数执行
37             for handler in self.__handlers[event.type_]:
38                 handler(event)
39
40     def Start(self):
41         """启动"""
42         # 将事件管理器设为启动
43         self.__active = True
44         # 启动事件处理线程
45         self.__thread.start()
46
47     def Stop(self):
48         """停止"""
49         # 将事件管理器设为停止
50         self.__active = False
51         # 等待事件处理线程退出
52         self.__thread.join()
53
54     def AddEventListener(self, type_, handler):
55         """绑定事件和监听器处理函数"""
56         # 尝试获取该事件类型对应的处理函数列表，若无则创建
57         try:
58             handlerList = self.__handlers[type_]
59         except KeyError:
60             handlerList = []
61
62         self.__handlers[type_] = handlerList
63         # 若要注册的处理器不在该事件的处理器列表中，则注册该事件
64         if handler not in handlerList:
65             handlerList.append(handler)
66
67     def RemoveEventListener(self, type_, handler):
68         """移除监听器的处理函数"""
69         #读者自己试着实现
70
71     def SendEvent(self, event):
72         """发送事件，向事件队列中存入事件"""
73         self.__eventQueue.put(event)
74
75 """事件对象"""
76 class Event:
77     def __init__(self, type_=None):
78         self.type_ = type_      # 事件类型
79         self.dict = {}          # 字典用于保存具体的事件数据

test.py

 1 # -*- encoding: UTF-8 -*-
 2
 3 from threading import *
 4 from EventManager import *
 5 import time
 6
 7 #事件名称  新文章
 8 EVENT_ARTICAL = "Event_Artical"
 9
10
11 #事件源 公众号
12 class PublicAccounts:
13     def __init__(self,eventManager):
14         self.__eventManager = eventManager
15
16     def WriteNewArtical(self):
17         #事件对象，写了新文章
18         event = Event(type_=EVENT_ARTICAL)
19         event.dict["artical"] = u‘如何写出更优雅的代码\n‘
20         #发送事件
21         self.__eventManager.SendEvent(event)
22         print(u‘公众号发送新文章‘)
23
24
25 #监听器 订阅者
26 class Listener:
27     def __init__(self,username):
28         self.__username = username
29
30     #监听器的处理函数 读文章
31     def ReadArtical(self,event):
32         print(u‘%s 收到新文章‘ % self.__username)
33         print(u‘正在阅读新文章内容：%s‘  % event.dict["artical"])
34
35
36 """测试函数"""
37 def test():
38     listner1 = Listener("thinkroom") #订阅者1
39     listner2 = Listener("steve")#订阅者2
40
41     eventManager = EventManager()
42
43     #绑定事件和监听器响应函数(新文章)
44     eventManager.AddEventListener(EVENT_ARTICAL, listner1.ReadArtical)
45     eventManager.AddEventListener(EVENT_ARTICAL, listner2.ReadArtical)
46     eventManager.Start()
47
48     publicAcc = PublicAccounts(eventManager)
49     while True:
50         publicAcc.WriteNewArtical()
51         time.sleep(2)
52
53 if __name__ == ‘__main__‘:
54     test()

原文地址：https://www.cnblogs.com/gundan/p/8313396.html