如何查询Celery路由队列中等待的任务并按ID撤销重发？

问题原因 #

celery inspect scheduled/reserved 只能查询已被worker节点接收但尚未执行的任务，而你在Redis中看到的任务还存放在broker的队列里（未被worker取走），所以inspect无法查到。要实现需求，需要直接操作Redis中的队列数据。

实现步骤 #

1. 连接Redis并遍历目标队列 #

Celery用Redis做broker时，队列以 queue:QUEUE_NAME 的Redis列表存储，每个列表项是序列化的任务消息。首先遍历你定义的所有队列，定位目标任务：

import redis
import json
import base64
from celery import Celery

# 初始化Redis连接（和原Celery应用用相同配置）
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# 定义要检查的队列列表
target_queue_prefix = "QUEUE_PREFIX"
queue_names = [f"{target_queue_prefix}.TA", f"{target_queue_prefix}.TB", f"{target_queue_prefix}.TC"]
target_task_id = "你要查找的任务ID"

found_task_info = None

2. 解析任务消息并匹配任务ID #

逐个读取队列中的消息，反序列化后检查任务ID。注意Celery默认会把任务参数用base64编码后放在body字段中：

for queue_name in queue_names:
    redis_queue_key = f"queue:{queue_name}"
    queue_len = redis_client.llen(redis_queue_key)
    
    for idx in range(queue_len):
        # 读取队列中第idx个元素（不修改原队列）
        msg_bytes = redis_client.lindex(redis_queue_key, idx)
        msg_data = json.loads(msg_bytes)
        
        # 匹配任务ID（properties中的id与任务内部id一致）
        if msg_data["properties"]["id"] == target_task_id:
            # 解码body获取args和kwargs
            body_decoded = base64.b64decode(msg_data["body"]).decode("utf-8")
            task_payload = json.loads(body_decoded)
            
            found_task_info = {
                "args": task_payload["args"],
                "kwargs": task_payload["kwargs"],
                "original_msg": msg_bytes,
                "queue_key": redis_queue_key,
                "queue_name": queue_name,
                "task_name": queue_name.split(".")[-1]  # 提取TA/TB/TC
            }
            break
    if found_task_info:
        break

3. 撤销旧任务并发送更新后的新任务 #

找到目标任务后，从Redis队列中移除旧任务，再用原参数（更新kwargs）发送新任务：

if not found_task_info:
    print("未找到目标任务")
else:
    # 1. 移除旧任务（根据原始消息内容删除，确保唯一性）
    redis_client.lrem(found_task_info["queue_key"], 0, found_task_info["original_msg"])
    
    # 2. 更新kwargs
    updated_kwargs = found_task_info["kwargs"].copy()
    updated_kwargs["需要修改的键"] = "新值"  # 替换为你的更新逻辑
    
    # 3. 初始化Celery应用（和原应用配置完全一致）
    celery = Celery(main='MY_APP', config_source=你的配置文件路径)
    
    # 4. 发送新任务到原队列
    celery.send_task(
        f"你的任务模块.{found_task_info['task_name']}",  # 例如 tasks.TA
        args=found_task_info["args"],
        kwargs=updated_kwargs,
        queue=found_task_info["queue_name"]
    )
    print("旧任务已撤销，新任务已发送")

注意事项 #

• 序列化兼容性：如果你的Celery用msgpack而非json序列化，需将json.loads替换为msgpack.unpackb，并调整base64解码后的处理逻辑。
• 原子性保障：遍历队列时可能出现worker同时取走任务的情况，可通过Redis事务或临时锁避免操作冲突。
• 配置一致性：独立应用的Celery配置（broker地址、序列化方式等）必须与原应用完全一致，否则发送新任务会失败。

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