如何在Celery任务中调用Python3.5异步函数？可行性与实现方法

1. 在Celery任务中使用Python异步函数是否合理？ #

当然可以，但得用对方法。Celery本身是基于同步执行模型设计的，但如果你的异步函数是处理I/O密集型操作（比如你这里的Redis连接池操作），在Celery任务里调用异步代码完全合理——它能帮你更高效地利用等待I/O的时间，提升任务的执行效率。只是要注意，Celery Worker是同步运行的，不能直接像调用普通函数那样使用异步函数，必须做一层适配处理。

2. 如何在Celery任务中调用Python协程函数？ #

你碰到的"'coroutine' object has no attribute XXX method'"错误，核心原因是：直接调用async def定义的函数时，返回的是协程对象，而不是函数执行后的实际结果（比如你的Redis连接池）。你必须主动运行这个协程，才能拿到想要的对象。

结合你使用的Python 3.5版本，这里给你两种可行的解决办法：

方法一：手动创建并运行事件循环（适配Python 3.5） #

因为Python 3.7才推出asyncio.run()，3.5版本需要手动处理事件循环：

import asyncio
from your_module import Cache

@app.task
def check_data_task():
    # 创建新的事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    # 运行协程直到完成，拿到实际的Redis连接池
    cache = loop.run_until_complete(Cache.get_pool())
    # 用完记得关闭循环
    loop.close()
    # 接下来就可以正常操作cache对象了
    ...

方法二：如果后续升级到Python 3.7+，可以用更简洁的方式 #

等你升级Python版本后，就能用asyncio.run()大幅简化代码：

import asyncio
from your_module import Cache

@app.task
def check_data_task():
    cache = asyncio.run(Cache.get_pool())
    # 正常操作cache对象即可
    ...

额外提示 #

• 如果你的Celery Worker要处理大量这类异步任务，频繁创建销毁事件循环会有性能损耗，可以考虑在Worker启动时初始化一个全局事件循环，复用它来处理所有任务的异步调用。
• 确保你的aioredis版本和Python 3.5兼容，避免出现版本不匹配的问题。

内容的提问来源于stack exchange，提问作者Hadi Farhadi