如何通过Task ID在Celery中查询任务状态（Django环境）

一、终端查询任务状态失败 #

问题原因 #

1. 命令中Celery实例名称错误：你的Celery app名为coutoEditor，但命令中用了proj，无法匹配正确的任务实例。
2. Task ID存在输入错误：你提供的ID末尾多了pe，不符合UUID规范。
3. 用错Celery命令：inspect query_task仅查询活跃/正在运行的任务，若任务已完成，该命令无法返回结果。

解决步骤 #

1. 使用正确的Celery实例名称执行命令：
   • 查询活跃任务（任务未完成时）：

     celery -A coutoEditor inspect query_task e9f6c8f0-fec9-4ae8-a8c6-cf8c8451d4f8
     

   • 查询任意状态的任务结果（无论是否完成）：

     celery -A coutoEditor result e9f6c8f0-fec9-4ae8-a8c6-cf8c8451d4f8
     

2. 确保输入的Task ID格式正确，UUID应为32位字符加4个连字符（如e9f6c8f0-fec9-4ae8-a8c6-cf8c8451d4f8）。
3. 执行命令前，确保Celery Worker处于运行状态：

   celery -A coutoEditor worker -l info
   

二、Redis直接查询任务状态无效 #

问题原因 #

Celery将任务元数据存储在Redis特定格式的键下，未使用正确键名则无法查询到数据。

解决步骤 #

1. 打开Redis命令行客户端：

   redis-cli
   

2. 使用Celery默认的任务元数据键格式查询：

   GET celery-task-meta-e9f6c8f0-fec9-4ae8-a8c6-cf8c8451d4f8
   

   返回的JSON结果包含status（任务状态：PENDING/SUCCESS/FAILURE等）和result（任务返回值）字段。
3. 验证Redis连接配置：确保settings.py中的CELERY_BROKER_URL和CELERY_RESULT_BACKEND与Redis实例的地址、端口、密码（若有）完全一致。

三、添加bind=True触发参数错误 #

问题原因 #

给@shared_task添加bind=True时，Celery会自动将任务实例对象作为第一个参数传入任务函数，但原函数未接收该参数，导致参数不匹配。

解决步骤 #

修改tasks.py中的任务定义，将self作为第一个参数：

@shared_task(bind=True)
def speed_up_vid_task(self, input_path, speed_factor, start, end):
    # 原函数逻辑保持不变，可通过self访问任务实例属性（如self.request.id获取当前任务ID）
    start = convert_to_sec(start)
    end = convert_to_sec(end)
    # ... 剩余代码 ...

额外优化建议 #

你的views.py中使用result_vid.get()会阻塞请求直到任务完成，易导致HTTP超时（尤其视频处理任务耗时较长）。建议改为返回Task ID，前端通过轮询或异步方式查询状态：

class speed_up_video(APIView):
    def post(self,request):
        video_url = request.data["video_url"]
        speed_factor = request.data["speed_factor"]
        start = request.data["start"]
        end = request.data["end"]
        result_vid = speed_up_vid_task.delay(video_url, speed_factor, start, end)
        return Response({"task_id": result_vid.id})

新增查询任务状态的API：

from celery.result import AsyncResult

class task_status(APIView):
    def get(self, request, task_id):
        result = AsyncResult(task_id)
        return Response({
            "task_id": task_id,
            "status": result.status,
            "result": result.result if result.status == "SUCCESS" else None
        })

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