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></coroutine>，下游拿不到真实值
• 若在 task 内手动调用 asyncio.run()，会在每次任务中新建并关闭 loop，开销大、无法复用连接（如 aiohttp session、aioredis pool），还可能触发 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()
  </p>

• 更高效的做法是复用 event loop + executor，在 worker 启动时初始化一次：

  import asyncio
  from celery import Celery
  <p>app = Celery('tasks', broker='redis://localhost')</p><div class="aritcle_card flexRow artxards">
  											<div class="artcardd flexRow">
  												<a class="aritcle_card_img" rel="nofollow" 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" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
  												<div class="aritcle_card_info flexColumn">
  													<a rel="nofollow" href="/xiazai/skill8910" title="361SCAN security check your skill python scripts" class="overflowclass">361SCAN security check your skill python scripts</a>
  													<p class="overflowclass">扫描并分析已安装技能，用于查看指定技能目录的名称、描述和详情，或列出所有已安装的技能。</p>
  												</div>
  												<a rel="nofollow" href="/xiazai/skill8910" title="361SCAN security check your skill python scripts" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
  												</a>
  											</div>
  										</div><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)))</code>

• 注意：不要在 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></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!。

Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！