Celery+Django+Redis架构下同任务多次执行及worker监控问题咨询

一、Celery任务重复执行问题修复 #

核心原因分析 #

你遇到的重复执行问题90%以上是Redis作为Broker的默认机制导致的：Celery从Broker取走任务后，若在默认visibility_timeout（1小时）内没有返回执行完成的ACK确认，Broker会判定任务执行失败，将任务重新投递给其他Worker，就会出现同一站点被多次执行的情况。另外你提供的代码本身存在两处明显bug也会触发重试：

1. post_data函数中md变量未初始化就直接做拼接运算，会抛出NameError异常
2. 异常捕获块为空，未做错误处理也未终止任务，会导致任务超时触发重发

修复步骤 #

• 调整Broker超时配置，在Django的settings.py中添加如下配置，将超时时间设置为大于你任务最长执行时间的值：

CELERY_BROKER_TRANSPORT_OPTIONS = {
    'visibility_timeout': 3 * 3600  # 设为3小时，可根据实际情况调整
}

• 修复代码bug：
  1. post_data函数开头初始化md变量：md = ""
  2. 补全异常捕获逻辑，不需要重试的场景可以直接标记任务完成，或者添加重试次数限制
• 给任务添加幂等校验，执行前先在Redis中写入执行标记，避免重复执行：
  在task_post_data遍历站点的逻辑中新增判断：

import redis
r = redis.Redis(host='localhost', port=6379, db=0)
for website in website_list:
    # 检查是否已经执行过
    if r.get(f"post_task:{website}"):
        continue
    # 写入标记，过期时间设为24小时
    r.setex(f"post_task:{website}", 24*3600, "1")
    # 原有执行逻辑

• 关闭不必要的自动重试，修改任务装饰器：

@app.task(bind=True, name="task_post_data", retry=False)
# 若需要重试可设置最大重试次数：retry_max=2, retry_backoff=60

二、Worker与任务监控方法 #

1. 命令行监控（Celery自带工具） #

• 查看所有Worker当前正在处理的任务：
  celery -A django_backend inspect active
• 查看Worker总数、每个Worker的处理统计数据：
  celery -A django_backend inspect stats
• 查看所有待执行的排队任务：
  celery -A django_backend inspect reserved

2. 代码内集成监控 #

可以在Django业务代码中直接调用Celery的Inspect接口获取监控数据：

from celery.task.control import inspect
celery_inspect = inspect()
# 获取所有活跃任务，返回字典的key为Worker名称，value为该Worker的任务列表
active_task_map = celery_inspect.active()
# 获取在线Worker数量
worker_count = len(active_task_map.keys()) if active_task_map else 0

3. 可视化监控 #

使用Celery官方配套的Flower工具即可实现Web端可视化监控，支持查看Worker在线状态、任务执行耗时、成功率、失败栈等信息，使用方法如下：

1. 安装Flower：pip install flower
2. 启动监控服务：celery -A django_backend flower
3. 访问http://本机IP:5555即可查看监控界面

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