如何通过Celery AsyncResult获取任务重试次数？

嘿，我之前也遇到过这个问题——确实，Celery的AsyncResult默认不会暴露任务的重试次数，不过有几个实用的办法可以解决，让你能在Flask页面上展示这个数据：

这是最直接的方式，在任务触发重试前，把当前的重试次数通过update_state存入任务的元数据（meta）中，之后就能通过AsyncResult读取到。

示例任务代码：

from celery import shared_task

@shared_task(bind=True)
def my_retry_task(self):
    try:
        # 这里写你的核心任务逻辑
        raise ValueError("模拟任务执行失败，触发重试")
    except Exception as e:
        # 把当前重试次数存入任务元数据
        self.update_state(meta={'retries': self.request.retries})
        # 触发重试，设置重试间隔（比如5秒）
        self.retry(exc=e, countdown=5)

对应的Flask端点代码：

from flask import Flask, jsonify
from celery.result import AsyncResult

app = Flask(__name__)

@app.route('/task/<task_id>')
def get_task_status(task_id):
    task_result = AsyncResult(task_id)
    # 从元数据中提取重试次数，默认值设为0
    retry_count = task_result.info.get('retries', 0) if task_result.info else 0
    
    return jsonify({
        'task_id': task_id,
        'status': task_result.status,
        'retry_count': retry_count,
        'result': task_result.result if task_result.successful() else None
    })

如果有多个任务需要追踪重试次数，写重复的update_state代码太麻烦，可以自定义一个任务基类，利用Celery的on_retry钩子自动记录重试次数：

from celery import Task

class RetryTrackingTask(Task):
    def on_retry(self, exc, task_id, args, kwargs, einfo):
        # 重试触发时自动更新元数据
        self.update_state(task_id=task_id, meta={'retries': self.request.retries})
        # 调用父类的on_retry方法，保证原有逻辑正常执行
        super().on_retry(exc, task_id, args, kwargs, einfo)

# 后续任务直接继承这个基类
@shared_task(base=RetryTrackingTask, bind=True)
def another_retry_task(self):
    try:
        # 任务逻辑
        raise IOError("模拟IO错误，触发重试")
    except Exception as e:
        self.retry(exc=e, countdown=10)

这样所有继承RetryTrackingTask的任务，在重试时都会自动把重试次数存入元数据，Flask端点的读取逻辑和方案1一致。

如果你的Celery用Redis作为结果后端，还可以直接从Redis中读取任务的原始请求数据——Celery会把任务的请求上下文（包括retries）存在celery-task-meta-<task_id>这个键里。

示例代码：

import redis
import json
from flask import Flask, jsonify
from celery.result import AsyncResult

app = Flask(__name__)
# 初始化Redis连接（根据你的配置调整参数）
redis_client = redis.Redis(host='localhost', port=6379, db=0)

@app.route('/task/<task_id>')
def get_task_status(task_id):
    task_result = AsyncResult(task_id)
    retry_count = 0
    
    # 从Redis获取任务元数据
    meta_key = f'celery-task-meta-{task_id}'
    task_meta_raw = redis_client.get(meta_key)
    
    if task_meta_raw:
        task_meta = json.loads(task_meta_raw)
        # 从request字段中提取重试次数
        retry_count = task_meta.get('request', {}).get('retries', 0)
    
    return jsonify({
        'task_id': task_id,
        'status': task_result.status,
        'retry_count': retry_count,
        'result': task_result.result if task_result.successful() else None
    })

这个方法不需要修改任务代码，但依赖于你使用的结果后端（这里是Redis），如果是其他后端（比如数据库），需要调整读取逻辑。

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