gpt4 book ai didi

python模拟事件触发机制详解

转载 作者:qq735679552 更新时间:2022-09-28 22:32:09 24 4
gpt4 key购买 nike

CFSDN坚持开源创造价值,我们致力于搭建一个资源共享平台,让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章python模拟事件触发机制详解由作者收集整理,如果你对这篇文章有兴趣,记得点赞哟.

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

EventManager.py 。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
# -*- 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 。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
# -*- 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()

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

原文链接:https://www.cnblogs.com/gundan/archive/2018/01/18/8313396.html 。

最后此篇关于python模拟事件触发机制详解的文章就讲到这里了,如果你想了解更多关于python模拟事件触发机制详解的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。

24 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com