Celery（Redis为Broker）任务重复堆积致内存飙升问题求助

核心问题总结 #

当前基于Django REST框架使用Celery+Redis处理回调任务时，出现以下异常：

• 内存每1分40秒增长约48MB
• 短时间内大量重复接收同一ID的send_callback_task任务，且无执行/失败日志
• 触发QoS: Disabled: prefetch_count exceeds 65535警告
• 服务器重启后恢复大量未确认消息

已尝试移除任务重试机制、为POST请求添加5秒超时，问题仍未解决。

排查与修复步骤 #

1. 调整Celery Worker预取数（prefetch_count） #

16核服务器默认的预取数会随核心数放大，容易触发QoS禁用警告，进而导致任务重复分发、未确认消息堆积。

修改配置：

# 在Django settings.py中添加
CELERY_WORKER_PREFETCH_MULTIPLIER = 1

或者启动Worker时直接指定：

celery -A project worker -l info --prefetch-multiplier=1

该配置限制Worker进程每次仅预取1个任务，处理完成后再取下一个，从根源避免预取数超限，减少未确认消息积压。

2. 修复任务代码的异常处理漏洞 #

当前代码存在异常未正确捕获、抛出的问题，可能导致任务无声崩溃，Broker误以为任务未被处理而重复分发。

修改后的任务代码：

import requests
from requests.exceptions import RequestException
from celery import shared_task
import logging

logger = logging.getLogger(__name__)

@shared_task(bind=True)
def send_callback_task(self, url: str, data):
    try:
        response = requests.post(url, json=data, timeout=5)
        logger.info({"msg": "Callback sent!", "url": url, "status_code": response.status_code})

        if response.status_code not in (200, 201, 202):
            logger.info({"msg": "Callback failed!", "url": url, "status_code": response.status_code})
            raise RequestException(f"Callback failed with status {response.status_code}")
        # 显式关闭请求连接，避免连接池泄漏
        response.close()
    except RequestException as e:
        logger.error({"msg": "Callback request error", "url": url, "error": str(e)})
        raise
    except Exception as e:
        # 捕获所有未预期异常，避免任务无声退出
        logger.error({"msg": "Unexpected error in callback task", "url": url, "error": str(e)})
        raise

• 明确导入并处理RequestException
• 添加全局异常捕获，记录详细错误日志
• 显式关闭请求连接，避免内存泄漏
• 确保异常被正确抛出，让Celery标记任务为失败，阻止Broker重复分发同ID任务

3. 优化Worker资源配置 #

16核默认生成16个Worker进程，结合CELERY_WORKER_MAX_TASKS_PER_CHILD = 100的配置，若任务存在内存泄漏，仍会导致内存持续增长。

调整配置：

# settings.py
CELERY_WORKER_MAX_TASKS_PER_CHILD = 50  # 减少子进程处理任务数，加快内存回收
CELERY_WORKER_CONCURRENCY = 8  # 根据业务压力调整进程数，避免资源过载

4. 检查Redis Broker状态 #

• 执行redis-cli INFO stats查看unacked_messages、pending_keys指标，确认未确认消息来源
• 检查Redis的maxmemory配置，避免内存不足导致消息异常
• 执行redis-cli KEYS "*celery*"清理过期任务元数据（生产环境需谨慎操作）

5. 验证任务触发逻辑 #

在调用apply_async的代码中添加日志，记录任务ID与触发时间，排查是否存在重复触发同一任务的情况。

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