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></coroutine>，而不是实际数据 - 现象：
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 artxards">
<div class="artcardd flexRow">
<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill9386" title="python学习"><img
src="https://img.php.cn/upload/skill/001/246/273/6adaf20e695c2374.jpg" alt="python学习" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
<div class="aritcle_card_info flexColumn">
<a rel="nofollow" href="/xiazai/skill9386" title="python学习" class="overflowclass">python学习</a>
<p class="overflowclass">Python 编程学习助手 skill，面向零基础用户。 当用户表达以下意图时触发： - 学习 Python 基础概念（变量、数据类型、循环、函数、类等） - 请求 Python 练习题或测验 - 请求审查或调试 Python 代码 - 请求 Python 项目实战指导（爬虫、数据分析、Web 开发等入门项目） - 泛化表述如"教我 Python""我想学编程""Python 怎么写""帮我学 Python" 此 skill</p>
</div>
<a rel="nofollow" href="/xiazai/skill9386" title="python学习" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
</a>
</div>
</div><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))</code>
-
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、连接池）几乎必然出错。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
