Django应用使用redis-cli查询Celery队列长度异常求助

问题原因与解决方法 #

核心原因1：队列名称存在统一前缀 #

你在Celery配置的broker_transport_options中大概率配置了queue_name_prefix参数，该参数会给所有队列的Redis key加上统一前缀，你直接查询队列名haystack当然匹配不到实际的key。
你可以打开你的Celery配置文件查找以下配置段：

CELERY_BROKER_TRANSPORT_OPTIONS = {
    # 类似如下配置就是前缀
    "queue_name_prefix": "your_proj_prefix:",
}

如果存在该配置，查询队列长度时需要加上前缀，示例如下：

redis-cli -n 0 -h ${MY_REDIS_HOST} -p 6379 llen your_proj_prefix:haystack

核心原因2：任务已被worker预取到unacked队列 #

Celery worker默认开启预取机制（prefetch_multiplier默认值为4），会提前拉取一定数量的任务到本地的unacked队列暂存等待执行，这部分任务已经从原始队列的list key中移除，所以你直接查询原始队列的llen返回0，但实际仍有任务待执行。
你看到的inspect scheduled返回的待执行任务，大部分都属于预取到worker本地的未确认任务。

核心原因3：使用了优先级队列 #

如果你给自定义队列配置了优先级参数，Celery会为每个优先级生成单独的Redis key，格式为队列名\x06优先级数值，不会直接使用你配置的队列名作为key，自然查询不到。

更准确的队列长度统计方案 #

如果你需要统计全量待执行任务用于自动扩缩容，不建议直接查询Redis key，直接调用Celery原生接口统计的结果更准确，可参考如下Python代码集成到你的Django项目中：

from celery import Celery

# 替换为你项目的celery实例
app = Celery('my-app')

def calc_queue_total_length(queue_name: str) -> int:
    # 1. 统计broker中尚未被worker预取的任务数
    with app.connection_for_read() as conn:
        pending_in_broker = conn.default_channel.client.llen(queue_name)
    
    # 2. 统计所有worker预取后尚未执行的任务数（包含active、reserved、scheduled三类）
    inspect_client = app.control.inspect()
    active_tasks = inspect_client.active() or {}
    reserved_tasks = inspect_client.reserved() or {}
    scheduled_tasks = inspect_client.scheduled() or {}
    
    pending_in_worker = 0
    # 统计运行中、预取待运行的任务
    for task_list in list(active_tasks.values()) + list(reserved_tasks.values()):
        pending_in_worker += len([
            t for t in task_list 
            if t["delivery_info"]["routing_key"] == queue_name
        ])
    # 统计定时/延迟待触发的任务
    for task_list in scheduled_tasks.values():
        pending_in_worker += len([
            t for t in task_list 
            if t["request"]["delivery_info"]["routing_key"] == queue_name
        ])
    
    return pending_in_broker + pending_in_worker

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