Python 实现事件监听

缘起

最近的项目需要用到事件监听机制，Google 了下，Python 似乎没有专门用于这方面的包。参考大牛发的 Snippet 写了个事件监听类，方便以后调用。

事件类

首先需要有自己的事件类，定义如下两个 Demo：

class Event1:
    """
    Event 1
    """
    value = "this is a event, <Event 1>"
class Event2:
    """
    Event 2
    """
    value = "this is a event, <Event 2>"

事件管理、分发

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
"""
Event Manager，负责任务的分发
"""
from queue import Queue, Empty
from multiprocessing.dummy import Pool
class EventManager:
    """
    事件分发
    """
    def __init__(self):
        """
        初始化
        """
        self.__event_handler = {}
        self.__pool = Pool(5)
    def add_event_handler(self, event_type, handler):
        """
        注册 event handler
        :param event_type: 事件类型
        :type event_type: object
        :param handler: Event Handler
        :type handler: function
        """
        self.__event_handler.setdefault(event_type, []).append(handler)
    def dispatch_event(self, event):
        """
        分发事件
        :param event: 事件
        :type event: Event
        :return: 是否分发成功
        :rtype: bool
        """
        try:
            handlers = self.__event_handler.get(event.__class__, [])
            for handler in handlers:
                self.__pool.apply_async(
                    func=handler,
                    args=(event,),
                    callback=self.__handler_callback
                )
            return True
        except:
            return False
    def __handler_callback(self, result):
        """
        进程回调
        :param result: 返回值
        :type result: str
        """
        print(result)

由上而下，类中共有 4 个方法。

_init_

self.__event_handler： 用于注册 handler

self.__pool： 以线程池的形式进行任务调度

add_event_handler

注册 event-handler，将 handler 保存至 self.__event_handler 中，以供后期分发事件。

dispatch_event

分发事件，做法是根据事件类型（event.__class__）从 self.__event_handler 中获取其 handler，依次将 handler(event) 添加至线程池中，callback 用于处理 handler 的返回结果，如不需要可以自行去掉。

__handler_callback

用于处理 handler 的返回结果，如不需要可以自行去掉。

测试

def handler1(event):
    """
    处理 Event1 的 handler
    """
    return "I'm handler1, monitored an event. Its value: {value}".format(value=event.value)
def handler2(event):
    """
    处理 Event2 的 handler
    """
    return "I'm handler2, monitored an event. Its value: {value}".format(value=event.value)
# 实例化 EventManager
em = EventManager()
# 注册 handler
em.add_event_handler(Event1, handler1)
em.add_event_handler(Event2, handler2)
# 分发事件
e1 = Event1()
e2 = Event2()
em.dispatch_event(e1)
em.dispatch_event(e2)

References

Simple event dispatcher (Python recipe)

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

# -----------------------------------------------------------------------------
# Event and EventDispatcher classes
# -----------------------------------------------------------------------------
    
class Event( object ):
    """
    Generic event to use with EventDispatcher.
    """
    
    def __init__(self, event_type, data=None):
        """
        The constructor accepts an event type as string and a custom data
        """
        self._type = event_type
        self._data = data
        
    @property 
    def type(self):
        """
        Returns the event type
        """
        return self._type
        
    @property
    def data(self):
        """
        Returns the data associated to the event
        """
        return self._data


class EventDispatcher( object ):
    """
    Generic event dispatcher which listen and dispatch events
    """
    
    def __init__(self):
        self._events = dict()
        
    def __del__(self):
        """
        Remove all listener references at destruction time
        """
        self._events = None
    
    def has_listener(self, event_type, listener):
        """
        Return true if listener is register to event_type
        """
        # Check for event type and for the listener
        if event_type in self._events.keys():
            return listener in self._events[ event_type ]
        else:
            return False
        
    def dispatch_event(self, event):
        """
        Dispatch an instance of Event class
        """
        # Dispatch the event to all the associated listeners 
        if event.type in self._events.keys():
            listeners = self._events[ event.type ]
            
            for listener in listeners:
                listener( event )
        
    def add_event_listener(self, event_type, listener):
        """
        Add an event listener for an event type
        """
        # Add listener to the event type
        if not self.has_listener( event_type, listener ):
            listeners = self._events.get( event_type, [] )
        
            listeners.append( listener )
            
            self._events[ event_type ] = listeners
    
    def remove_event_listener(self, event_type, listener):
        """
        Remove event listener.
        """
        # Remove the listener from the event type
        if self.has_listener( event_type, listener ):
            listeners = self._events[ event_type ]
            
            if len( listeners ) == 1:
                # Only this listener remains so remove the key
                del self._events[ event_type ]
                
            else:
                # Update listeners chain
                listeners.remove( listener )
                
                self._events[ event_type ] = listeners


# ------------------------------------------------------------------------------
# Events and Dispatcher example
#
# In this example we create a simple event MyEvent with only two event types,
# ASK and RESPOND, and two classes: WhoAsk, which send AKS event and listen for
# the RESPOND event, and WhoRespond, which listen for ASK events and send back
# a RESPOND event
# -----------------------------------------------------------------------------

class MyEvent( Event ):
    """
    When subclassing Event class the only thing you must do is to define
    a list of class level constants which defines the event types and the 
    string associated to them
    """
    
    ASK     = "askMyEvent"
    RESPOND = "respondMyEvent"


class WhoAsk( object ):
    """
    First class which ask who is listening to it
    """
    def __init__(self, event_dispatcher):
        # Save a reference to the event dispatch
        self.event_dispatcher = event_dispatcher
        
        # Listen for the RESPOND event type
        self.event_dispatcher.add_event_listener( 
            MyEvent.RESPOND, self.on_answer_event 
        )
        
    def ask(self):
        """
        Dispatch the ask event
        """
        print ">>> I'm instance {0}. Who are listening to me ?".format( self )

        self.event_dispatcher.dispatch_event( 
            MyEvent( MyEvent.ASK, self ) 
        )
        
    def on_answer_event(self, event):
        """
        Event handler for the RESPOND event type
        """
        print "<<< Thank you instance {0}".format( event.data )
        

class WhoRespond( object ):
    """
    Second class who respond to ASK events
    """
    def __init__(self, event_dispatcher):
        # Save event dispatcher reference
        self.event_dispatcher = event_dispatcher
        
        # Listen for ASK event type
        self.event_dispatcher.add_event_listener( 
            MyEvent.ASK, self.on_ask_event 
        )
        
    def on_ask_event(self, event):
        """
        Event handler for ASK event type
        """
        self.event_dispatcher.dispatch_event( 
            MyEvent ( MyEvent.RESPOND, self ) 
        )


if __name__ == "__main__":
    # Create and instance of event dispatcher
    dispatcher = EventDispatcher()
    
    # Create an instance of WhoAsk class and two instance of WhoRespond class
    who_ask = WhoAsk( dispatcher )
    who_responde1 = WhoRespond( dispatcher )
    who_responde2 = WhoRespond( dispatcher )
    
    # WhoAsk ask :-)
    who_ask.ask()