python多線程與多進程異步事件框架

時間 2019-11-08

原文原文鏈接

多線程簡單實現

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import logging
import queue
import threading
from concurrent.futures import ThreadPoolExecutor


# 任務：事件
def func_a(a, b):
    return a + b
def func_b(a, b):
    return a * b
def func_c(a, b, c):
    return a * b - c
# 回調函數
def handle_result1(result):
    print(type(result), result)
def handle_result2(result):
    print(type(result), result)
def handle_result3(result):
    print(type(result), result)


class EventEngine(object):
    # 初始化事件事件驅動引擎
    def __init__(self):
        # 保存事件列表:異步任務隊列
        self.__eventQueue = queue.Queue()
        # 引擎開關
        self.__active = False
        # 事件處理字典{'event1': [handler1,handler2] , 'event2':[handler3, ...,handler4]}
        self.__handlers = {}
        # 事件引擎主進程
        self.__Thread = threading.Thread(target=self.task_queue_consumer)
        # 事件處理線程池
        self.__thread_pool = ThreadPoolExecutor(max_workers=5)
        # 線程處理存儲
        self.__thread_Pool = []

    #註冊事件
    def register(self,event, callback, *args, **kwargs):
        Event = {
            'function': event,
            'callback': callback,
            'args': args,
            'kwargs': kwargs
        }
        self.__handlers[event] = Event

    #註銷事件
    def unregister(self,event):
        if(self.__handlers[event]):
            del self.__handlers[event]


    #提交事件
    def sendevent(self,event):
        if ( event in self.__handlers.keys()):
            self.__eventQueue.put(self.__handlers[event])


    # 開啓事件引擎
    def start(self):
        self.__active = True
        self.__Thread.start()

    # 暫停事件引擎
    def stop(self):
        self.__active = False

    # 暫停後開始
    def restart(self):
        self.__active = True

    # 關閉事件引擎
    def close(self):
        pass

    # 開啓事件循環
    def task_queue_consumer(self):
        """
        異步任務隊列
        """
        while(1):
            while self.__active:
                if (self.__eventQueue.empty() == False):
                    try:
                        task = self.__eventQueue.get()
                        function = task.get('function')
                        callback = task.get('callback')
                        args = task.get('args')
                        kwargs = task.get('kwargs')
                        try:
                            if callback:
                                thread = self.__thread_pool.submit(callback,function(*args, **kwargs))
                                self.__thread_Pool.append(thread)
                                # callback(function(*args, **kwargs))
                        except Exception as ex:
                            if callback:
                                callback(ex)
                        finally:
                            self.__eventQueue.task_done()
                    except Exception as ex:
                        logging.warning(ex)



if __name__ == '__main__':
    import time
    #初始化多線程異步框架
    Engine = EventEngine()
    #啓動
    Engine.start()
    #註冊回調函數
    Engine.register(func_a, handle_result1, 1, 2)
    Engine.register(func_b, handle_result2, 1, 2)
    Engine.register(func_c, handle_result3, 1, 2, 3)
    #提交事件
    Engine.sendevent(func_a)
    Engine.sendevent(func_b)
    Engine.sendevent(func_c)
    time.sleep(2)
    Engine.stop()
    Engine.restart()
    Engine.sendevent(func_b)
    Engine.sendevent(func_c)
    # for i in range(100):
    #     Engine.sendevent(func_a)

多進程實現

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