Celery任务中不能直接await,因其Worker运行在无事件循环的同步进程里;正确做法是用asyncio.run()在任务内新建并管理事件循环。

Celery 本身不是异步框架,它的 Worker 默认运行在同步进程里,直接在 @task 函数中写 await 会静默失败或抛 RuntimeWarning: coroutine 'xxx' was never awaited —— 这是绝大多数人踩坑的第一步。
为什么 Celery 任务里不能直接 await?
Celery Worker 启动的是标准 Python 进程(fork 或 spawn),不自带事件循环。即使你的主程序用了 asyncio,Worker 进程里也没有 asyncio.get_event_loop() 可用。
- 现象:任务返回
<coroutine object fetch_data at 0x...>,而不是实际数据 - 现象:
await asyncio.sleep(1)看似执行了,但后续代码不走、日志不打、结果为空 - 根本原因:Celery 不是为协程设计的调度器;它只负责把函数调用“扔进进程”,不管里面是不是协程
正确做法:在任务内部用 asyncio.run() 启动新事件循环
必须把异步逻辑封装成普通函数,再用 asyncio.run() 包一层。不能依赖外部事件循环,也不能复用全局 loop。
from celery import Celery
import asyncio
import aiohttp
<p>app = Celery('tasks', broker='redis://localhost:6379/0')</p><div class="aritcle_card flexRow">
<div class="artcardd flexRow">
<a class="aritcle_card_img" href="/xiazai/skill9386" title="python学习"><img
src="https://img.php.cn/upload/skill/001/246/273/6adaf20e695c2374.jpg" alt="python学习" ></a>
<div class="aritcle_card_info flexColumn">
<a href="/xiazai/skill9386" title="python学习">python学习</a>
<p>Python 编程学习助手 skill,面向零基础用户。 当用户表达以下意图时触发: - 学习 Python 基础概念(变量、数据类型、循环、函数、类等) - 请求 Python 练习题或测验 - 请求审查或调试 Python 代码 - 请求 Python 项目实战指导(爬虫、数据分析、Web 开发等入门项目) - 泛化表述如"教我 Python""我想学编程""Python 怎么写""帮我学 Python" 此 skill</p>
</div>
<a href="/xiazai/skill9386" title="python学习" 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><p>async def _fetch_url(url: str) -> dict:
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return {"status": resp.status, "url": url}</p><p>@app.task
def fetch_url_task(url: str) -> dict:</p><h1>✅ 正确:每次任务都新建并关闭事件循环</h1><pre class="brush:php;toolbar:false;"><code>return asyncio.run(_fetch_url(url))-
asyncio.run()是安全边界:它确保 loop 创建、运行、关闭全过程可控 - 不要在任务外定义
aiohttp.ClientSession()并传入 —— session 不能跨asyncio.run()生命周期 - 避免在任务里调用
asyncio.get_event_loop()或asyncio.new_event_loop(),多 worker 场景下易报RuntimeError: There is no current event loop
并发瓶颈:单个 Worker 进程仍只能串行跑 asyncio.run()
哪怕你用了 asyncio,默认 Celery Worker 是单线程同步执行模型 —— 它不会自动把多个任务并发 dispatch 到同一个事件循环里。
- 效果:一个任务内用
asyncio.gather()可以并发 IO,但多个fetch_url_task.delay()调用仍是排队执行 - 解法一:启动多个 Worker 进程(
-c 4)来真正并行 - 解法二:改用
eventlet或gevent模式(需安装celery[eventlet],启动加-P eventlet),让单个 Worker 支持协程并发 - 注意:
eventlet不兼容某些 C 扩展(如psycopg2的二进制版),生产前务必验证
和 FastAPI / Flask 异步路由共存时的关键隔离点
如果你的 Web 框架(如 FastAPI)本身在运行事件循环,千万别把 Celery 任务注册成 async def 然后试图用 create_task() 提交到那个 loop —— Celery Worker 根本不接入那个 loop。
- 错误示范:
@app.task(async def send_email(...): ...)→ 语法错误 - 错误示范:在 FastAPI 路由里
asyncio.create_task(celery_task.delay())→ 没意义,delay()本身是同步调用 - 正确姿势:Web 层调用
task.delay()或task.apply_async()即可,这是纯同步操作;异步逻辑只收束在任务函数体内
最常被忽略的一点:Celery + asyncio 不是“让 Celery 变成异步”,而是“在同步任务壳子里安全跑异步逻辑”。所有协程生命周期必须严格限制在单次 asyncio.run() 内,跨任务复用资源(比如全局 session、连接池)几乎必然出错。

















