python - 哪些 Python 包提供独立的事件系统？

Question

我知道pydispatcher，但是 Python 必须有其他与事件相关的包。

哪些库可用？

我对作为大型框架一部分的事件管理器不感兴趣，我宁愿使用可以轻松扩展的小型准系统解决方案。

Answer 1

PyPI 包

截至 2022 年 1 月，这些是 PyPI 上可用的与事件相关的包，按最近的发布日期排序。

• 即插即用 1.0.0：2021 年 8 月
• pymitter 0.3.1 : 2021 年 6 月
• zope.event 4.5.0 : 2020 年 9 月
• python-dispatch 0.1.31：2020 年 8 月
• RxPy3 1.0.1：2020 年 6 月
• 路易 2.0：2019 年 9 月
• PyPubSub 4.0.3：2019 年 1 月
• pyevent调度程序 0.2.3a0：2018
• 巴士兰 0.0.5: 2018
• PyPyDispatcher 2.1.2：2017
• 轴 0.0.7：2016
• 信号灯 1.4：2015
• PyDispatcher 2.0.5：2015 年
• 调度员 1.0：2012
• py 通知 0.3.1：2008

还有更多

有很多库可供选择，使用非常不同的术语（事件、信号、处理程序、方法分派、钩子……）。

我试图保持对上述软件包的概述，以及此处答案中提到的技术。

首先，一些术语...

观察者模式

事件系统最基本的风格是“处理程序方法包”，它是观察者模式的简单实现。

基本上，处理程序方法（可调用对象）存储在一个数组中，并在事件“触发”时被调用。

发布-订阅

Observer 事件系统的缺点是您只能在实际的 Event 对象（或处理程序列表）上注册处理程序。因此，在注册时，该事件已经存在。

这就是存在第二种事件系统样式的原因： 发布-订阅模式。在这里，处理程序不是在事件对象（或处理程序列表）上注册，而是在中央调度程序上注册。此外，通知者只与调度员交谈。听什么或发布什么由“信号”决定，它只不过是一个名称（字符串）。

中介者模式

可能也很有趣：中介者模式。

挂钩

“挂钩”系统通常用于应用程序插件的上下文中。应用程序包含固定的集成点（挂钩），每个插件都可以连接到该挂钩并执行某些操作。

其他“事件”

注意：threading.Event不是上述意义上的“事件系统”。这是一个线程同步系统，其中一个线程一直等待，直到另一个线程“发出”事件对象的“信号”。

网络消息传递库也经常使用术语“事件”；有时这些在概念上是相似的；有时不是。它们当然可以跨越线程、进程和计算机边界。参见例如 pyzmq、pymq、 Twisted、Tornado、 gevent、eventlet。

弱引用

在 Python 中，持有对方法或对象的引用可确保它不会被垃圾收集器删除。这可能是可取的，但也可能导致内存泄漏：链接的处理程序永远不会被清理。

一些事件系统使用弱引用而不是常规引用来解决这个问题。

关于各种图书馆的一些话

观察者式事件系统：

• zope.event展示了它的工作原理（参见Lennart 的回答）。注意：此示例甚至不支持处理程序参数。
• LongPoke 的“可调用列表”实现表明，这样的事件系统可以通过子类化非常简单地实现list。
• Felk 的变体EventHook还确保了被调用者和调用者的签名。
• spassig 的 EventHook（Michael Foord 的事件模式）是一个简单的实现。
• Josip 的 Valued Lessons Event 类基本相同，只是用 aset代替了 alist来存放包，而且__call__它们都是合理的补充。
• PyNotify在概念上类似，并且还提供了变量和条件的附加概念（“变量更改事件”）。主页无法正常工作。
• axel基本上是一个处理程序包，具有与线程、错误处理相关的更多功能......
• python-dispatch要求偶数源类从pydispatch.Dispatcher.
• buslane是基于类的，支持单个或多个处理程序并促进广泛的类型提示。
• Pithikos 的Observer/Event是一种轻量级设计。

发布订阅库：

• blinker有一些漂亮的功能，例如自动断开连接和基于发件人的过滤。
• PyPubSub是一个稳定的包，并承诺“有助于调试和维护主题和消息的高级功能”。
• pymitter是 Node.js EventEmitter2 的 Python 端口，提供命名空间、通配符和 TTL。
• PyDispatcher似乎强调多对多发布等方面的灵活性。支持弱引用。
• louie是一个重新设计的 PyDispatcher，应该“在各种环境中”工作。
• pypydispatcher基于（你猜对了......）PyDispatcher 并且也适用于 PyPy。
• django.dispatch是一个重写的 PyDispatcher，“接口更有限，但性能更高”。
• pyeventdispatcher基于 PHP 的 Symfony 框架的事件调度器。
• 调度程序是从 django.dispatch 中提取的，但已经相当老了。
• Cristian Garcia 的EventManger是一个非常简短的实现。

其他：

• pluggy包含一个由pytest插件使用的钩子系统。
• RxPy3实现了 Observable 模式并允许合并事件、重试等。
• Qt 的信号和插槽可从PyQt 或PySide2 获得。它们在同一线程中使用时用作回调，或者在两个不同线程之间用作事件（使用事件循环）。Signals 和 Slots 的局限性在于它们只能在派生自QObject.

Answer 2

我一直这样做：

class Event(list):
    """Event subscription.

    A list of callable objects. Calling an instance of this will cause a
    call to each item in the list in ascending order by index.

    Example Usage:
    >>> def f(x):
    ...     print 'f(%s)' % x
    >>> def g(x):
    ...     print 'g(%s)' % x
    >>> e = Event()
    >>> e()
    >>> e.append(f)
    >>> e(123)
    f(123)
    >>> e.remove(f)
    >>> e()
    >>> e += (f, g)
    >>> e(10)
    f(10)
    g(10)
    >>> del e[0]
    >>> e(2)
    g(2)

    """
    def __call__(self, *args, **kwargs):
        for f in self:
            f(*args, **kwargs)

    def __repr__(self):
        return "Event(%s)" % list.__repr__(self)

然而，就像我见过的其他一切一样，没有为此自动生成的 pydoc，也没有签名，这真的很糟糕。

Answer 3

我们使用 Michael Foord 在他的Event Pattern中建议的 EventHook ：

只需将 EventHooks 添加到您的类中：

class MyBroadcaster()
    def __init__():
        self.onChange = EventHook()

theBroadcaster = MyBroadcaster()

# add a listener to the event
theBroadcaster.onChange += myFunction

# remove listener from the event
theBroadcaster.onChange -= myFunction

# fire event
theBroadcaster.onChange.fire()

我们添加了从对象中删除所有侦听器到 Michaels 类的功能，并最终得到以下结果：

class EventHook(object):

    def __init__(self):
        self.__handlers = []

    def __iadd__(self, handler):
        self.__handlers.append(handler)
        return self

    def __isub__(self, handler):
        self.__handlers.remove(handler)
        return self

    def fire(self, *args, **keywargs):
        for handler in self.__handlers:
            handler(*args, **keywargs)

    def clearObjectHandlers(self, inObject):
        for theHandler in self.__handlers:
            if theHandler.im_self == inObject:
                self -= theHandler

Answer 4

我使用zope.event。这是你能想象到的最赤裸裸的骨头。:-) 事实上，这里是完整的源代码：

subscribers = []

def notify(event):
    for subscriber in subscribers:
        subscriber(event)

请注意，例如，您不能在进程之间发送消息。它不是一个消息系统，只是一个事件系统，仅此而已。

Answer 5

我在Valued Lessons上找到了这个小脚本。它似乎具有我所追求的恰到好处的简单性/功率比。Peter Thatcher 是以下代码的作者（未提及许可）。

class Event:
    def __init__(self):
        self.handlers = set()

    def handle(self, handler):
        self.handlers.add(handler)
        return self

    def unhandle(self, handler):
        try:
            self.handlers.remove(handler)
        except:
            raise ValueError("Handler is not handling this event, so cannot unhandle it.")
        return self

    def fire(self, *args, **kargs):
        for handler in self.handlers:
            handler(*args, **kargs)

    def getHandlerCount(self):
        return len(self.handlers)

    __iadd__ = handle
    __isub__ = unhandle
    __call__ = fire
    __len__  = getHandlerCount

class MockFileWatcher:
    def __init__(self):
        self.fileChanged = Event()

    def watchFiles(self):
        source_path = "foo"
        self.fileChanged(source_path)

def log_file_change(source_path):
    print "%r changed." % (source_path,)

def log_file_change2(source_path):
    print "%r changed!" % (source_path,)

watcher              = MockFileWatcher()
watcher.fileChanged += log_file_change2
watcher.fileChanged += log_file_change
watcher.fileChanged -= log_file_change2
watcher.watchFiles()

Answer 6

这是一个应该可以正常工作的最小设计。你所要做的就是简单地继承Observer一个类，然后observe(event_name, callback_fn)用来监听一个特定的事件。每当该特定事件在代码中的任何位置（即Event('USB connected')）被触发时，相应的回调就会触发。

class Observer():
    _observers = []
    def __init__(self):
        self._observers.append(self)
        self._observed_events = []
    def observe(self, event_name, callback_fn):
        self._observed_events.append({'event_name' : event_name, 'callback_fn' : callback_fn})


class Event():
    def __init__(self, event_name, *callback_args):
        for observer in Observer._observers:
            for observable in observer._observed_events:
                if observable['event_name'] == event_name:
                    observable['callback_fn'](*callback_args)

例子：

class Room(Observer):
    def __init__(self):
        print("Room is ready.")
        Observer.__init__(self) # DON'T FORGET THIS
    def someone_arrived(self, who):
        print(who + " has arrived!")

# Observe for specific event
room = Room()
room.observe('someone arrived',  room.someone_arrived)

# Fire some events
Event('someone left',    'John')
Event('someone arrived', 'Lenard') # will output "Lenard has arrived!"
Event('someone Farted',  'Lenard')

Answer 7

我创建了一个EventManager类（最后的代码）。语法如下：

#Create an event with no listeners assigned to it
EventManager.addEvent( eventName = [] )

#Create an event with listeners assigned to it
EventManager.addEvent( eventName = [fun1, fun2,...] )

#Create any number event with listeners assigned to them
EventManager.addEvent( eventName1 = [e1fun1, e1fun2,...], eventName2 = [e2fun1, e2fun2,...], ... )

#Add or remove listener to an existing event
EventManager.eventName += extra_fun
EventManager.eventName -= removed_fun

#Delete an event
del EventManager.eventName

#Fire the event
EventManager.eventName()

这是一个例子：

def hello(name):
    print "Hello {}".format(name)
    
def greetings(name):
    print "Greetings {}".format(name)

EventManager.addEvent( salute = [greetings] )
EventManager.salute += hello

print "\nInitial salute"
EventManager.salute('Oscar')

print "\nNow remove greetings"
EventManager.salute -= greetings
EventManager.salute('Oscar')

输出：

最初的敬礼
问候奥斯卡
你好奥斯卡

现在删除问候
你好奥斯卡

事件管理器代码：

class EventManager:
    
    class Event:
        def __init__(self,functions):
            if type(functions) is not list:
                raise ValueError("functions parameter has to be a list")
            self.functions = functions
            
        def __iadd__(self,func):
            self.functions.append(func)
            return self
            
        def __isub__(self,func):
            self.functions.remove(func)
            return self
            
        def __call__(self,*args,**kvargs):
            for func in self.functions : func(*args,**kvargs)
            
    @classmethod
    def addEvent(cls,**kvargs):
        """
        addEvent( event1 = [f1,f2,...], event2 = [g1,g2,...], ... )
        creates events using **kvargs to create any number of events. Each event recieves a list of functions,
        where every function in the list recieves the same parameters.
        
        Example:
        
        def hello(): print "Hello ",
        def world(): print "World"
        
        EventManager.addEvent( salute = [hello] )
        EventManager.salute += world
        
        EventManager.salute()
        
        Output:
        Hello World
        """
        for key in kvargs.keys():
            if type(kvargs[key]) is not list:
                raise ValueError("value has to be a list")
            else:
                kvargs[key] = cls.Event(kvargs[key])
        
        cls.__dict__.update(kvargs)

Answer 8

你可以看看pymitter ( pypi )。它是一种小型单文件（~250 loc）方法“提供命名空间、通配符和 TTL”。

这是一个基本示例：

from pymitter import EventEmitter

ee = EventEmitter()

# decorator usage
@ee.on("myevent")
def handler1(arg):
   print "handler1 called with", arg

# callback usage
def handler2(arg):
    print "handler2 called with", arg
ee.on("myotherevent", handler2)

# emit
ee.emit("myevent", "foo")
# -> "handler1 called with foo"

ee.emit("myotherevent", "bar")
# -> "handler2 called with bar"

Answer 9

我对 Longpoke 的简约方法做了一个变体，它也确保了被调用者和调用者的签名：

class EventHook(object):
    '''
    A simple implementation of the Observer-Pattern.
    The user can specify an event signature upon inizializazion,
    defined by kwargs in the form of argumentname=class (e.g. id=int).
    The arguments' types are not checked in this implementation though.
    Callables with a fitting signature can be added with += or removed with -=.
    All listeners can be notified by calling the EventHook class with fitting
    arguments.

    >>> event = EventHook(id=int, data=dict)
    >>> event += lambda id, data: print("%d %s" % (id, data))
    >>> event(id=5, data={"foo": "bar"})
    5 {'foo': 'bar'}

    >>> event = EventHook(id=int)
    >>> event += lambda wrong_name: None
    Traceback (most recent call last):
        ...
    ValueError: Listener must have these arguments: (id=int)

    >>> event = EventHook(id=int)
    >>> event += lambda id: None
    >>> event(wrong_name=0)
    Traceback (most recent call last):
        ...
    ValueError: This EventHook must be called with these arguments: (id=int)
    '''
    def __init__(self, **signature):
        self._signature = signature
        self._argnames = set(signature.keys())
        self._handlers = []

    def _kwargs_str(self):
        return ", ".join(k+"="+v.__name__ for k, v in self._signature.items())

    def __iadd__(self, handler):
        params = inspect.signature(handler).parameters
        valid = True
        argnames = set(n for n in params.keys())
        if argnames != self._argnames:
            valid = False
        for p in params.values():
            if p.kind == p.VAR_KEYWORD:
                valid = True
                break
            if p.kind not in (p.POSITIONAL_OR_KEYWORD, p.KEYWORD_ONLY):
                valid = False
                break
        if not valid:
            raise ValueError("Listener must have these arguments: (%s)"
                             % self._kwargs_str())
        self._handlers.append(handler)
        return self

    def __isub__(self, handler):
        self._handlers.remove(handler)
        return self

    def __call__(self, *args, **kwargs):
        if args or set(kwargs.keys()) != self._argnames:
            raise ValueError("This EventHook must be called with these " +
                             "keyword arguments: (%s)" % self._kwargs_str())
        for handler in self._handlers[:]:
            handler(**kwargs)

    def __repr__(self):
        return "EventHook(%s)" % self._kwargs_str()

Answer 10

如果我在 pyQt 中编写代码，我使用 QT 套接字/信号范式，同样适用于 django

如果我正在执行异步 I/OI，请使用本机选择模块

如果我使用 SAX python 解析器，我将使用 SAX 提供的事件 API。所以看起来我是底层 API 的受害者:-)

也许你应该问问自己你对事件框架/模块有什么期望。我个人的偏好是使用 QT 的 Socket/Signal 范例。更多信息可以在这里找到

Answer 11

这是另一个需要考虑的模块。对于要求更高的应用程序，这似乎是一个可行的选择。

Py-notify 是一个 Python 包，提供了实现观察者编程模式的工具。这些工具包括信号、条件和变量。

信号是发出信号时调用的处理程序列表。条件基本上是布尔变量，加上条件状态变化时发出的信号。可以使用标准逻辑运算符（非和等）将它们组合成复合条件。与条件不同，变量可以保存任何 Python 对象，而不仅仅是布尔值，但它们不能组合。

Answer 12

如果你想做更复杂的事情，比如合并事件或重试，你可以使用 Observable 模式和实现它的成熟库。https://github.com/ReactiveX/RxPY。Observables 在 Javascript 和 Java 中非常常见，并且在一些异步任务中使用起来非常方便。

from rx import Observable, Observer


def push_five_strings(observer):
        observer.on_next("Alpha")
        observer.on_next("Beta")
        observer.on_next("Gamma")
        observer.on_next("Delta")
        observer.on_next("Epsilon")
        observer.on_completed()


class PrintObserver(Observer):

    def on_next(self, value):
        print("Received {0}".format(value))

    def on_completed(self):
        print("Done!")

    def on_error(self, error):
        print("Error Occurred: {0}".format(error))

source = Observable.create(push_five_strings)

source.subscribe(PrintObserver())

输出：

Received Alpha
Received Beta
Received Gamma
Received Delta
Received Epsilon
Done!

Answer 13

如果您需要跨进程或网络边界工作的事件总线，您可以尝试PyMQ。它目前支持发布/订阅、消息队列和同步 RPC。默认版本在 Redis 后端之上运行，因此您需要一个正在运行的 Redis 服务器。还有一个用于测试的内存后端。您也可以编写自己的后端。

import pymq

# common code
class MyEvent:
    pass

# subscribe code
@pymq.subscriber
def on_event(event: MyEvent):
    print('event received')

# publisher code
pymq.publish(MyEvent())

# you can also customize channels
pymq.subscribe(on_event, channel='my_channel')
pymq.publish(MyEvent(), channel='my_channel')

初始化系统：

from pymq.provider.redis import RedisConfig

# starts a new thread with a Redis event loop
pymq.init(RedisConfig())

# main application control loop

pymq.shutdown()

免责声明：我是这个库的作者

Answer 14

另一个方便的包是events。它封装了事件订阅和事件触发的核心，感觉就像是语言的“自然”部分。它看起来类似于 C# 语言，它提供了一种方便的方式来声明、订阅和触发事件。从技术上讲，事件是可以附加回调函数（事件处理程序）的“槽”——称为订阅事件的过程。

# Define a callback function
def something_changed(reason):
    print "something changed because %s" % reason

# Use events module to create an event and register one or more callback functions
from events import Events
events = Events()
events.on_change += something_changed

触发事件时，将依次调用所有附加的事件处理程序。要触发事件，请在插槽上执行调用：

events.on_change('it had to happen')

这将输出：

'something changed because it had to happen'

更多文档可以在github repo或文档中找到。

Answer 15

你可以试试buslane模块。

该库使基于消息的系统的实现更容易。它支持命令（单个处理程序）和事件（0 或多个处理程序）方法。Buslane 使用 Python 类型注释来正确注册处理程序。

简单的例子：

from dataclasses import dataclass

from buslane.commands import Command, CommandHandler, CommandBus


@dataclass(frozen=True)
class RegisterUserCommand(Command):
    email: str
    password: str


class RegisterUserCommandHandler(CommandHandler[RegisterUserCommand]):

    def handle(self, command: RegisterUserCommand) -> None:
        assert command == RegisterUserCommand(
            email='john@lennon.com',
            password='secret',
        )


command_bus = CommandBus()
command_bus.register(handler=RegisterUserCommandHandler())
command_bus.execute(command=RegisterUserCommand(
    email='john@lennon.com',
    password='secret',
))

要安装 buslane，只需使用 pip：

$ pip install buslane

Answer 16

前段时间我写了一个可能对你有用的库。它允许您拥有本地和全局侦听器、多种不同的注册方式、执行优先级等。

from pyeventdispatcher import register

register("foo.bar", lambda event: print("second"))
register("foo.bar", lambda event: print("first "), -100)

dispatch(Event("foo.bar", {"id": 1}))
# first second

看看pyeventdispatcher