如何在Celery中获取所有任务的ID及状态（无需已知任务ID）

Celery的result_backend确实是获取任务状态的核心入口，但不同后端的实现逻辑不同，以下是两种常用后端的具体操作方法：

一、Redis作为Result Backend #

Celery会将任务元数据以celery-task-meta-<task_id>为key存储在Redis中，直接扫描这类key即可批量获取任务信息：

基础实现代码 #

from celery import Celery
import redis
import json

app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

# 连接Redis实例
redis_conn = redis.Redis(host='localhost', port=6379, db=0)

# 扫描所有任务元数据key（大数量场景建议用scan替代keys）
cursor = 0
task_keys = []
while True:
    cursor, keys = redis_conn.scan(cursor, match='celery-task-meta-*', count=100)
    task_keys.extend(keys)
    if cursor == 0:
        break

# 解析所有任务信息
all_tasks = []
for key in task_keys:
    task_meta = redis_conn.get(key)
    if task_meta:
        meta_data = json.loads(task_meta)
        all_tasks.append({
            'task_id': meta_data['task_id'],
            'status': meta_data['status'],
            'result': meta_data.get('result'),
            'date_done': meta_data.get('date_done')
        })

# 输出结果
for task in all_tasks:
    print(f"任务ID: {task['task_id']}, 状态: {task['status']}, 完成时间: {task.get('date_done')}")

关键提示 #

• 用scan替代keys：keys命令会阻塞Redis实例，在任务数量较多时必须改用scan进行迭代扫描
• 任务过期：如果配置了result_expires，过期的任务元数据会被自动删除，无法被获取

二、SQLAlchemy作为Result Backend #

当使用数据库作为结果后端时，Celery会自动创建celery_taskmeta表，直接查询该表即可获取所有任务状态：

实现代码 #

from celery import Celery
from sqlalchemy import create_engine, text

app = Celery('tasks', broker='pyamqp://guest@localhost//', backend='db+mysql://user:password@localhost/db_name')

# 创建数据库连接
engine = create_engine(app.conf.result_backend)

with engine.connect() as conn:
    # 查询所有任务记录
    result = conn.execute(text("SELECT task_id, status, result, date_done FROM celery_taskmeta"))
    all_tasks = [
        {
            'task_id': row.task_id,
            'status': row.status,
            'result': row.result,
            'date_done': row.date_done
        }
        for row in result
    ]

# 输出结果
for task in all_tasks:
    print(f"任务ID: {task['task_id']}, 状态: {task['status']}, 完成时间: {task.get('date_done')}")

通用注意事项 #

• 其他后端（如MongoDB、RabbitMQ）的存储结构不同，需要对应调整查询逻辑
• 生产环境中尽量避免全量扫描，可通过任务分组、自定义标记等方式缩小查询范围

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