详解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()

来源：http://www.jianshu.com/p/5e7786166157?utm_source=tuicool&utm_medium=referral