FastAPI+RabbitMQ+Celery代码排查：任务始终未就绪

核心问题现象 #

FastAPI的根接口可正常将任务发送至RabbitMQ队列并返回任务ID，但调用/api/result/{task_id}查询状态时，所有任务ID的task.ready()均返回False，无法获取执行结果。

具体错误排查与修复方案 #

1. 任务路由与Worker监听队列不匹配 #

• 问题分析：
  你在Celery配置中指定了任务路由：

  celery_app.conf.task_routes = {"celery_worker.test_celery": "test-queue"}
  

  但从Celery服务信息可见，Worker仅监听了默认的celery队列，未监听你配置的test-queue：

  [queues]
  .> celery           exchange=celery(direct) key=celery
  

  任务被发送到test-queue后，Worker没有消费该队列的任务，导致任务一直处于PENDING状态，task.ready()自然返回False。

• 修复方案：
  二选一即可：

  • 调整任务路由，将任务发送到Worker监听的celery队列：

    celery_app.conf.task_routes = {"celery_worker.test_celery": "celery"}
    

  • 启动Worker时指定监听test-queue，修改docker-compose.yml中celery_worker的command：

    command: celery -A main.celery_app worker --loglevel=INFO -Q celery,test-queue
    

2. AsyncResult未关联Celery实例 #

• 问题分析：
  在result接口中，你直接使用AsyncResult(task_id)初始化任务对象，但该方式没有关联到你配置好的celery_app实例，无法正确读取rpc://后端的任务状态数据。

• 修复方案：
  使用celery_app实例的AsyncResult方法来初始化：

  @app.get("/api/result/{task_id}")
  async def result(task_id: str):
      # 改为通过celery_app获取AsyncResult
      task = celery_app.AsyncResult(task_id)
      if not task.ready():
          return {"status": task.status}
      task_result = task.get()
      result = task_result.get("result")
      return {
          "task_id": str(task_id),
          "status": task_result.get("status"),
          "result": result,
      }
  

3. 任务ID返回格式错误 #

• 问题分析：
  在根接口中，你返回的任务ID是f"{task}"，这会返回类似celery@415bde516932-xxxxxx的字符串，而正确的任务ID应该是task.id。错误的ID格式会导致查询时无法匹配到真实任务，返回False。

• 修复方案：
  修改根接口的返回字段：

  @app.get("/{word}")
  async def root(word: str):
      task = celery_app.send_task("celery_worker.test_celery", args=[word])
      # 返回正确的task.id
      return {"message": "Word received", "id": task.id}
  

额外验证点 #

确保你的celery_worker.test_celery任务函数已正确定义，且Worker能正常导入该函数。如果任务函数不存在或导入失败，Worker会丢弃任务，也会导致状态异常。

内容的提问来源于stack exchange，提问作者Evgeny Romensky