Celery 任务函数不能直接 await，因其 worker 基于同步模型，不原生支持 async/await；正确做法是用 asyncio.to_thread() 或线程池执行异步逻辑，或改用 arq 等原生异步任务队列。
在 Python 异步环境中直接用 Celery 处理异步任务，本质上是行不通的——Celery 的 worker 是同步阻塞模型，不原生支持 async/await。强行在 task 函数里写 await 会报 RuntimeWarning: coroutine 'xxx' was never awaited，或导致事件循环混乱。
为什么 Celery 任务函数不能直接 await
Celery 的 task 执行器（如 pool=prefork）默认使用多进程，每个 worker 进程启动时没有运行中的 asyncio 事件循环；即使改用 pool=threads，线程内也未自动创建 loop，且 celery 的调用链（apply_async → run）不是 awaitable。
- 你写的
@app.task函数被当作普通同步 callable 注册，Celery 不会用asyncio.run()或loop.create_task()包装它 -
async def my_task(): ...被注册后，实际执行的是一个 coroutine object，而非运行结果——这会导致任务返回<coroutine object my_task at 0x...>，下游拿不到真实值 - 若在 task 内手动调用
asyncio.run()，会在每次任务中新建并关闭 loop，开销大、无法复用连接（如aiohttpsession、aioredispool），还可能触发RuntimeError: asyncio.run() cannot be called from a running event loop
正确做法：用 asyncio.to_thread() 或 loop.run_in_executor() 包裹异步逻辑
把真正耗时的异步 I/O 操作（如调用 aiohttp、aiomysql）放到线程池里执行，让 Celery 保持同步接口，但内部调度异步代码。这是目前最稳妥、兼容性最好的方案。
- Python ≥ 3.9 推荐用
asyncio.to_thread()：@app.task def fetch_data_task(url): import asyncio return asyncio.run(async_fetch(url)) <p>async def async_fetch(url): import aiohttp async with aiohttp.ClientSession() as session: async with session.get(url) as resp: return await resp.text() - 更高效的做法是复用 event loop + executor，在 worker 启动时初始化一次：
import asyncio from celery import Celery <p>app = Celery('tasks', broker='redis://localhost')</p><div class="aritcle_card flexRow"> <div class="artcardd flexRow"> <a class="aritcle_card_img" href="/xiazai/skill8910" title="361SCAN security check your skill python scripts"><img src="https://img.php.cn/upload/skill/000/000/081/179231049778675.jpg" alt="361SCAN security check your skill python scripts" ></a> <div class="aritcle_card_info flexColumn"> <a href="/xiazai/skill8910" title="361SCAN security check your skill python scripts">361SCAN security check your skill python scripts</a> <p>扫描并分析已安装技能，用于查看指定技能目录的名称、描述和详情，或列出所有已安装的技能。</p> </div> <a href="/xiazai/skill8910" title="361SCAN security check your skill python scripts" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span> </a> </div> </div><p><span>立即学习</span>“<a href="https://pan.quark.cn/s/00968c3c2c15" style="text-decoration: underline !important; color: blue; font-weight: bolder;" rel="nofollow" target="_blank">Python免费学习笔记（深入）</a>”；</p><h1>全局共享的 event loop 和 executor（仅限单线程 worker）</h1><p>_loop = None _executor = None</p><p>@app.task def fetch_url(url): global _loop, _executor if _loop is None: _loop = asyncio.new_event_loop() _executor = _loop.create_default_executor()</p><h1>在 executor 中运行协程</h1><pre class="brush:php;toolbar:false;"><code>return _loop.run_until_complete(_loop.create_task(async_fetch(url))) - 注意：不要在
fetch_url内反复调用asyncio.new_event_loop()，否则 session、连接池无法复用，会快速耗尽文件描述符
替代方案：放弃 Celery，改用 arq 或 selene
如果项目重度依赖异步生态（比如大量 async def service 层、需要 async contextmanager），硬套 Celery 反而增加心智负担。此时应换真正为 async 设计的任务队列。
-
arq原生支持async def task()，worker 自动管理 event loop，可直接await数据库/HTTP 客户端：from arq import create_pool from arq.connections import RedisSettings <p>async def my_job(ctx, url: str): async with aiohttp.ClientSession() as session: async with session.get(url) as r: return await r.text()</p><h1>注册任务、触发执行都天然 async</h1><p>
-
arq依赖 Redis，配置简单，但不支持 RabbitMQ；selene支持多种 broker，但生态较新、文档少 - 迁移成本：需重写任务注册方式、监控集成（
arq自带 Web UI，但 Metrics 需自行对接 Prometheus）
真正棘手的不是“怎么让 async 代码跑起来”，而是连接复用和生命周期管理——比如一个 aioredis.Redis 实例必须在同一个 loop 中创建和使用，跨线程或跨 loop 传递会出错。这点在 Celery 场景下极易被忽略，直到压测时出现大量 Connection closed 或 Task was destroyed but it is pending!。
