生产环境中Celery任务重复执行问题求助

核心问题分析 #

从你的日志和配置来看，几个关键异常点：

• Beat与Worker同进程启动，结合Django DatabaseScheduler易引发调度逻辑混乱
• 日志出现missed heartbeat，说明Worker与Redis Broker的连接存在波动，Broker可能误判Worker已死并重新分发任务
• 同一调度时间的send_out_chat_message任务被Beat多次触发，表明调度器可能重复读取了未标记为已执行的任务记录

具体排查与修复步骤 #

1. 拆分Beat与Worker进程 #

当前将Beat和Worker放在同一进程启动的方式，生产环境下极易干扰调度逻辑，建议拆分启动：

# 单独启动Beat进程
celery -A bot.celery beat --scheduler django --loglevel=info
# 单独启动Worker进程
celery -A bot.celery worker --loglevel=info --concurrency 4

同进程下Worker的fork操作会打乱Beat的定时轮询节奏，导致任务重复入队。

2. 优化Redis Broker连接稳定性 #

missed heartbeat是Worker与Redis连接异常的典型表现，需强化连接配置：

CELERY_BROKER_URL = "redis://your-internal-redis-url/0"
# 添加Redis连接池参数，提升稳定性
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'max_connections': 10,
    'socket_timeout': 30,
    'socket_connect_timeout': 30,
    'socket_keepalive': True,
}
CELERY_RESULT_BACKEND = "redis://your-internal-redis-url/1"

同时检查Redis实例是否存在内存不足、持久化异常重启等情况，避免Broker数据丢失或重复分发。

3. 修复DatabaseScheduler重复调度问题 #

日志中同一时间点的任务被多次触发，大概率是数据库调度器的任务记录未正确标记：

• 检查Django的django_celery_beat表，清理PeriodicTask和IntervalSchedule中的重复记录
• 动态创建任务时添加唯一约束，避免重复插入调度记录
• 考虑切换到RedisScheduler替代DatabaseScheduler，减少数据库依赖带来的延迟与重复：

celery -A bot.celery beat --scheduler celery.beat.RedisScheduler --loglevel=info

4. 给任务添加幂等性保障 #

即使解决调度和Broker问题，生产环境仍需通过幂等性避免业务影响：

• 为按钮触发类任务生成唯一标识（如用户ID+按钮ID+时间戳）
• 任务执行前先校验是否已执行过，示例代码：

from celery import shared_task
import redis

redis_client = redis.Redis.from_url("redis://your-internal-redis-url/2")

@shared_task
def send_out_chat_message(unique_task_key, ...):
    # 校验任务是否已执行
    if redis_client.get(f"task_done:{unique_task_key}"):
        return None
    # 执行任务业务逻辑
    ...
    # 标记任务已完成，设置过期时间避免内存占用
    redis_client.setex(f"task_done:{unique_task_key}", 3600, "1")

5. 配置Worker任务超时 #

并发进程若被长耗时任务阻塞，易引发心跳超时，需添加超时限制：

CELERY_TASK_TIME_LIMIT = 300  # 任务最长执行时间5分钟
CELERY_TASK_SOFT_TIME_LIMIT = 240  # 软超时提前发送终止信号

日志验证建议 #

开启DEBUG日志，更清晰追踪任务流转：

celery -A bot.celery worker --loglevel=debug --concurrency 4

同时监控Redis中celery队列的长度变化，确认是否存在任务重复入队情况。

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