将异步协程作为芹菜任务运行

Question

我希望能够在我的芹菜任务中使用异步代码。它适用于异步或龙卷风。我发现，我可以做这样的事情

from tornado.ioloop import IOLoop
from celery._state import _task_stack
from . import celery

class AsyncTask(celery.Task):
def __call__(self, *args, **kwargs):
_task_stack.push(self)
self.push_request(args=args, kwargs=kwargs)
try:
IOLoop.current().run_sync(lambda: self.run(*args, **kwargs))
finally:
self.pop_request()
_task_stack.pop()

然后像这样使用它

from .celery import celery
from tornado.httpclient import AsyncHTTPClient
@celery.task(base=AsyncTask)
async def test_async_celery_task(x, y):
result = await AsyncHTTPClient().fetch(request='https://google.com.ua')
print('Async IS OKAY: {}'.format(result))

或者，我可以直接在任务中使用run_sync，这不可取

我想知道是否可以执行上述操作，或者我应该在 worker 中启动事件循环并通过add_future启动我的任务。还有其他人做过类似的事情吗？我可以希望获得一些性能增强吗？

我需要这个，因为我需要能够使用来自项目其他部分的异步代码，例如数据库调用 ets

Answer 1

不是 ioloop 的正确使用使它变得复杂，而是让 Celery 知道它可以而且应该使用异步任务。Celery 与 Pools 一起运营，它可以在其中安排您的工作。池可以是线程化的，可以多处理，Celery 知道(通过配置(它有 X 个进程、Y 个线程、Z 个工作线程，其中有多少有工作或空闲。但是ATM，Celery在技术上能够接收和运行协程，但开箱即用无法计算它们以跟踪其中有多少人有工作或空闲。如果你想看看它是什么样子 - 有一个未来Celery's Asyncpool的开发代码。