python实现事件驱动_Python

本文实例为大家分享了python实现事件驱动的具体代码，供大家参考，具体内容如下

eventmanager事件管理类实现，大概就百来行代码左右。

									# 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 = {}   # 字典用于保存具体的事件数据

测试代码

									#-------------------------------------------------------------------

									# encoding: utf-8

									import sys

									from datetime import datetime

									from threading import *

									from eventmanager import *

									#事件名称 新文章

									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'公众号发送新文章\n'

									#监听器 订阅者

									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)

									 timer = timer(2, publicacc.writenewartical)

									 timer.start()

									if __name__ == '__main__':

									 test()