详解Python实现多进程异步事件驱动引擎

yipeiwu_com6年前Python基础

本文介绍了详解Python实现多进程异步事件驱动引擎,分享给大家,具体如下:

多进程异步事件驱动逻辑

逻辑

code

# -*- coding: utf-8 -*-

'''
author:    Jimmy
contact:   234390130@qq.com
file:     eventEngine.py
time:     2017/8/25 上午10:06
description: 多进程异步事件驱动引擎

'''

__author__ = 'Jimmy'


from multiprocessing import Process, Queue


class EventEngine(object):
  # 初始化事件事件驱动引擎
  def __init__(self):
    #保存事件列表
    self.__eventQueue = Queue()
    #引擎开关
    self.__active = False
    #事件处理字典{'event1': [handler1,handler2] , 'event2':[handler3, ...,handler4]}
    self.__handlers = {}
    #保存事件处理进程池
    self.__processPool = []
    #事件引擎主进程
    self.__mainProcess = Process(target=self.__run)


  #执行事件循环
  def __run(self):
    while self.__active:
      #事件队列非空
      if not self.__eventQueue.empty():
        #获取队列中的事件 超时1秒
        event = self.__eventQueue.get(block=True ,timeout=1)
        #执行事件
        self.__process(event)
      else:
        # print('无任何事件')
        pass


  #执行事件
  def __process(self, event):
    if event.type in self.__handlers:
      for handler in self.__handlers[event.type]:
        #开一个进程去异步处理
        p = Process(target=handler, args=(event, ))
        #保存到进程池
        self.__processPool.append(p)
        p.start()


  #开启事件引擎
  def start(self):
    self.__active = True
    self.__mainProcess.start()


  #暂停事件引擎
  def stop(self):
    """停止"""
    # 将事件管理器设为停止
    self.__active = False
    # 等待事件处理进程退出
    for p in self.__processPool:
      p.join()
    self.__mainProcess.join()


  #终止事件引擎
  def terminate(self):
    self.__active = False
    #终止所有事件处理进程
    for p in self.__processPool:
      p.terminate()
    self.__mainProcess.join()


  #注册事件
  def register(self, type, handler):
    """注册事件处理函数监听"""
    # 尝试获取该事件类型对应的处理函数列表,若无则创建
    try:
      handlerList = self.__handlers[type]
    except KeyError:
      handlerList = []
      self.__handlers[type] = handlerList

    # 若要注册的处理器不在该事件的处理器列表中,则注册该事件
    if handler not in handlerList:
      handlerList.append(handler)


  def unregister(self, type, handler):
    """注销事件处理函数监听"""
    # 尝试获取该事件类型对应的处理函数列表,若无则忽略该次注销请求
    try:
      handlerList = self.__handlers[type]

      # 如果该函数存在于列表中,则移除
      if handler in handlerList:
        handlerList.remove(handler)

      # 如果函数列表为空,则从引擎中移除该事件类型
      if not handlerList:
        del self.__handlers[type]
    except KeyError:
      pass


  def sendEvent(self, event):
    #发送事件 像队列里存入事件
    self.__eventQueue.put(event)


class Event(object):
  #事件对象
  def __init__(self, type =None):
    self.type = type
    self.dict = {}



#测试
if __name__ == '__main__':
  import time
  EVENT_ARTICAL = "Event_Artical"

  # 事件源 公众号
  class PublicAccounts:
    def __init__(self, eventManager):
      self.__eventManager = eventManager

    def writeNewArtical(self):
      # 事件对象,写了新文章
      event = Event(EVENT_ARTICAL)
      event.dict["artical"] = u'如何写出更优雅的代码\n'
      # 发送事件
      self.__eventManager.sendEvent(event)
      print(u'公众号发送新文章\n')


  # 监听器 订阅者
  class ListenerTypeOne:
    def __init__(self, username):
      self.__username = username

    # 监听器的处理函数 读文章
    def ReadArtical(self, event):
      print(u'%s 收到新文章' % self.__username)
      print(u'%s 正在阅读新文章内容:%s' % (self.__username, event.dict["artical"]))


  class ListenerTypeTwo:
    def __init__(self, username):
      self.__username = username

    # 监听器的处理函数 读文章
    def ReadArtical(self, event):
      print(u'%s 收到新文章 睡3秒再看' % self.__username)
      time.sleep(3)
      print(u'%s 正在阅读新文章内容:%s' % (self.__username, event.dict["artical"]))


  def test():
    listner1 = ListenerTypeOne("thinkroom") # 订阅者1
    listner2 = ListenerTypeTwo("steve") # 订阅者2

    ee = EventEngine()

    # 绑定事件和监听器响应函数(新文章)
    ee.register(EVENT_ARTICAL, listner1.ReadArtical)
    ee.register(EVENT_ARTICAL, listner2.ReadArtical)
    for i in range(0, 20):
      listner3 = ListenerTypeOne("Jimmy") # 订阅者X
      ee.register(EVENT_ARTICAL, listner3.ReadArtical)

    ee.start()

    #发送事件
    publicAcc = PublicAccounts(ee)
    publicAcc.writeNewArtical()

  test()

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持【听图阁-专注于Python设计】。

相关文章

Pandas实现dataframe和np.array的相互转换

网上找了半天 不是dataframe转化成array的就是array转化dataframe,所以这里给汇总一下,相互转换的python代如下: dataframe转化成array d...

python绘制BA无标度网络示例代码

python绘制BA无标度网络示例代码

如下所示: #Copyright (c)2017, 东北大学软件学院学生 # All rightsreserved #文件名称:a.py # 作 者:孔云 #问题描述: #问题分析...

python中时间转换datetime和pd.to_datetime详析

python中时间转换datetime和pd.to_datetime详析

前言 我们在python对数据进行操作时,经常会选取某一时间段的数据进行分析。这里为大家介绍两个我经常用到的用来选取某一时间段数据的函数:datetime( )和pd.to_dateti...

python之文件读取一行一行的方法

如下所示: f=file('a.txt') for eachline in f: print eachline 以上这篇python之文件读取一行一行的方法就是小编分享给大家的...

盘点提高 Python 代码效率的方法

第一招:蛇打七寸:定位瓶颈 首先,第一步是定位瓶颈。举个简单的栗子,一个函数可以从1秒优化到到0.9秒,另一个函数可以从1分钟优化到30秒,如果要花的代价相同,而且时间限制只能搞定一个,...