如何在Celery+Redis环境下检查特定参数的任务是否已调度/执行？

Celery本身没有内置按参数查询任务的功能，但结合Redis作为Broker和Result Backend，我们可以通过几种方式实现需求：

1. 解析Redis队列中的待调度任务 #

Redis作为Broker时，待执行的任务会存在指定队列（默认是celery）的列表中，每个消息是序列化后的任务数据。我们可以读取队列内容，反序列化后匹配参数：

import redis
from celery import Celery
from kombu.serialization import loads

# 初始化Celery和Redis客户端
app = Celery('tasks', broker='redis://localhost:6379/0')
redis_cli = redis.Redis(host='localhost', port=6379, db=0)

def has_pending_task(task_name, target_args):
    queue_name = 'celery'  # 替换为你的目标队列名
    # 取出队列中所有消息
    raw_messages = redis_cli.lrange(queue_name, 0, -1)
    
    for msg in raw_messages:
        # 反序列化消息，确保和Celery配置的序列化方式一致
        payload = loads(msg)
        # 匹配任务名和参数
        if payload['task'] == task_name and payload['args'] == target_args:
            return True
    return False

# 使用示例：检查tasks.add任务（参数1,2）是否在待调度队列
print(has_pending_task('tasks.add', (1, 2)))

• 注意：此方法仅能检测等待调度的任务，无法获取正在执行或已完成的任务状态
• 要保证loads使用的序列化方式和Celery的CELERY_TASK_SERIALIZER配置一致（默认是json）

2. 从Result Backend查询执行中/已完成的任务 #

如果你的Celery配置了Result Backend（比如Redis），可以遍历存储的任务元数据，过滤匹配参数的任务：

from celery.result import AsyncResult

def has_running_or_completed_task(task_name, target_args):
    # 获取Redis中所有任务元数据的键（格式为celery-task-meta-<task_id>）
    task_keys = redis_cli.keys('celery-task-meta-*')
    
    for key in task_keys:
        task_id = key.decode().split('-')[-1]
        result = AsyncResult(task_id, app=app)
        # 匹配任务名和参数，同时检查状态是否为待执行或执行中
        if (result.name == task_name 
            and result.args == target_args 
            and result.state in ('PENDING', 'STARTED')):
            return True
    return False

• 注意：如果任务量很大，这种遍历方式性能极低，仅适合小规模场景或临时排查
• 任务完成后（SUCCESS/FAILURE），如果不需要保留状态，建议及时清理，避免元数据过多

3. 自定义任务参数追踪（生产环境推荐） #

频繁按参数查询任务的话，最靠谱的方式是在调度任务时主动记录参数与Task ID的映射：

def schedule_task_with_track(task_name, args):
    # 调度任务
    task = app.send_task(task_name, args=args)
    # 用参数的哈希值作为键，存储Task ID（可根据需求设计键的格式）
    param_hash = hash(tuple(args))
    redis_key = f'track:{task_name}:{param_hash}'
    # 设置过期时间，避免无用数据堆积（比如任务完成后1小时过期）
    redis_cli.setex(redis_key, 3600, task.id)
    return task

def check_task_by_params(task_name, args):
    param_hash = hash(tuple(args))
    redis_key = f'track:{task_name}:{param_hash}'
    task_id = redis_cli.get(redis_key)
    if not task_id:
        return False
    # 查询任务状态
    result = AsyncResult(task_id.decode(), app=app)
    return result.state in ('PENDING', 'STARTED')

• 优势：性能最优，直接通过Redis键查询，无需遍历
• 注意：需要在任务完成后主动清理映射（比如在任务函数末尾删除对应的Redis键），或者依赖过期时间自动清理

总结 #

• 临时排查用：解析Redis队列（待调度任务）或遍历Result Backend（执行中任务）
• 生产环境用：自定义参数追踪机制，兼顾性能和可靠性

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