使用asyncio进行异步信号处理

Question

我的模型后处理使用 post_save 信号：

from django.core.signals import request_finished
from django.dispatch import receiver
from models import MyModel
from pipeline import this_takes_forever


@receiver(post_save, sender=MyModel)
def my_callback(sender, **kwargs):
    this_takes_forever(sender)

这个this_takes_forever例程涉及到IO操作，因此我想将其推迟以避免请求被阻塞太久。

我认为这是使用新的asyncio模块的绝佳用例。但是我很难理解整个过程。

我认为我应该能够像这样调整信号接收器：

@receiver(post_save, sender=MyModel)
def my_callback(sender, **kwargs):
    loop = asyncio.get_event_loop()
    loop.run_until_complete(this_takes_forever(sender))
    loop.close()

假设this_takes_forever也被适配成协程。

@coroutine
def this_takes_forever(instance):
    # do something with instance
    return instance

这听起来太神奇了，实际上它会因为一个AssertionError而停止：

AssertionError at /new/
There is no current event loop in thread 'Thread-1'.

我不知道在这种情况下应该从哪里开始循环。有人尝试过类似于这样的操作吗？

- tutuca

Answer 1

您的情况并不会获得任何好处：

@receiver(post_save, sender=MyModel)
def my_callback(sender, **kwargs):
    this_takes_forever(sender)

等于

@receiver(post_save, sender=MyModel)
def my_callback(sender, **kwargs):
    loop = asyncio.get_event_loop()
    loop.run_until_complete(this_takes_forever(sender))
    loop.close()

在执行时间方面，loop.run_until_complete等待this_takes_forever(sender)协程调用结束，因此你在第二种情况下和前一种情况一样获得同步调用。

关于AssertionError：您在多线程模式下启动Django应用程序，但是asyncio仅为主线程提供默认事件循环--您应该为每个需要调用asyncio代码的用户创建的线程注册新的事件循环。

但是，请再说一遍，asyncio无法解决您的特定问题，它与Django不兼容。

Django的标准方式是将长时间运行的代码延迟到celery任务中（请参见http://www.celeryproject.org/）。

- Andrew Svetlov

---

第一点没问题。也许这只是一个不好的例子，我会尝试重构以获得更清晰的代码。无论如何，我没有在任何多线程模式下使用django，只是运行默认的runserver，它只有一个线程。深入研究后发现，我需要连接一些非阻塞服务器，比如aiohttp。 - tutuca

据我所知，runserver 默认会创建一个新线程，在其中运行 Django 应用程序，这就是 autoreload 功能的工作原理。 - Andrew Svetlov

关于aiohttp--请注意，与Django或tornado.web相比，该库现在太低级了。我正在开发更加用户友好的界面，但工作还处于早期阶段。 - Andrew Svetlov

也许我没有表达清楚。我说的是使用aiohttp而不是内置的runserver来运行django，请看这里。 - tutuca