如何通过任务ID查询Celery任务的完成百分比？

嘿，这个需求很实用！Celery本身并没有内置像你示例里a4654.percentage()这样直接获取任务完成百分比的方法，但我们完全可以自己实现这个功能，下面给你两种靠谱的方案：

Celery允许任务在执行过程中主动更新自身状态，我们可以把进度信息塞进状态的元数据里，之后不管是通过任务实例还是任务ID，都能轻松拿到百分比。

先写一个带进度更新的任务示例：

from celery import Celery

# 初始化Celery应用，这里用Redis做 broker，你可以换成自己的配置
app = Celery('progress_tasks', broker='redis://localhost:6379/0')

@app.task(bind=True)  # 必须加bind=True，才能在任务内访问self对象
def long_running_task(self, total_work_steps):
    for current_step in range(total_work_steps):
        # 这里替换成你实际的任务逻辑
        process_single_step(current_step)
        
        # 计算当前进度百分比
        progress_percent = int((current_step + 1) / total_work_steps * 100)
        # 更新任务状态，把进度数据存在meta字段里
        self.update_state(
            state='PROGRESS',  # 自定义状态名，用来标识任务正在执行且有进度
            meta={
                'current': current_step + 1,
                'total': total_work_steps,
                'percentage': progress_percent
            }
        )
    return "任务执行完成！"

接下来是查询进度的方式：

# 场景1：你持有任务实例（比如刚调用delay后的返回值）
task_instance = long_running_task.delay(100)  # 假设总共有100步
if task_instance.state == 'PROGRESS':
    print(f"{task_instance.info['percentage']}%")
elif task_instance.state == 'SUCCESS':
    print("100%")

# 场景2：只有任务ID，需要通过ID查询
from celery.result import AsyncResult
task_result = AsyncResult('你的任务ID字符串', app=app)
if task_result.state == 'PROGRESS':
    print(f"{task_result.info['percentage']}%")
elif task_result.state == 'SUCCESS':
    print("100%")
elif task_result.state == 'PENDING':
    print("任务还未开始执行")

如果你不想自己写状态更新的逻辑，可以用现成的第三方库，比如celery-progress。它已经封装好了进度跟踪的功能，你只需要按照文档配置即可，但本质上还是基于Celery的状态更新机制实现的。不过如果项目不想引入额外依赖，方法一完全够用。

• 任务必须设置bind=True，否则无法在任务内部调用self.update_state()。
• 自定义状态名可以随便取，但用PROGRESS是行业惯例，更容易理解。
• 如果任务已经失败（状态为FAILURE），可以根据业务需求返回0%或者提示任务失败。

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