Flask+Celery定时第三方通知任务重复发送问题求助

我来帮你排查这个重复任务的坑——这种ETA任务重复触发的情况在Celery+Redis的组合里确实挺常见，尤其是涉及第三方服务的时候。咱们从根源到解决方案一步步拆解：

核心原因分析 #

你的问题里有几个关键线索：

• 仅出现在调用第三方服务的ETA任务中，数据库任务无此问题
• 本地和生产环境都存在
• 重复次数随机（10-20+条）

结合你的代码和配置，可能的原因包括：

1. 任务缺乏幂等性：即使任务被Celery重复调度，也没有机制阻止重复发送通知
2. ACK机制与Worker重启的冲突：你开启了CELERY_ACKS_LATE=True，如果Worker意外退出（比如Heroku dyno重启），未ACK的任务会被重新入队执行
3. 重复调度任务：schedule_event_main可能被多次触发，导致同一个ETA任务被多次提交
4. 第三方服务超时导致任务重试：邮件/SMS发送耗时较长，触发Celery的默认重试逻辑（即使你注释了self.retry()，全局配置可能仍生效）

针对性解决方案 #

1. 优先实现任务幂等性（最关键） #

这是解决重复通知的根本保障——即使任务被重复执行，也不会重复发送。我们可以用Redis做分布式锁，给每个发送任务加唯一标识：

from celery.utils.log import get_task_logger
from redis import Redis
import hashlib

logger = get_task_logger(__name__)
redis_client = Redis()

# 修改单个发送任务，加入幂等校验
@app.task(bind=True)
def schedule_individual_ratings_emails(self, guest_name, host, guest_email, url, guest_id, event_id):
    # 生成唯一任务标识（结合guest和event，确保每个通知只发一次）
    task_key = hashlib.md5(f"rating_email_{guest_id}_{event_id}".encode()).hexdigest()
    
    # 用Redis的setnx实现分布式锁，过期时间1小时防止死锁
    if redis_client.set(task_key, "processed", ex=3600, nx=True):
        try:
            email_rating(guest_name, host, guest_email, url)
            logger.info(f"成功发送评分邮件给 {guest_email} (活动ID: {event_id})")
        except Exception as e:
            logger.error(f"发送邮件失败 {guest_email}: {str(e)}", exc_info=e)
            # 发送失败时释放锁，允许后续重试
            redis_client.delete(task_key)
            # 可控重试：最多3次，每次间隔60秒
            self.retry(exc=e, max_retries=3, countdown=60)
    else:
        logger.info(f"邮件已发送过，跳过 {guest_email} (活动ID: {event_id})")

注意：调用这个任务时要传入guest_id和event_id，从EventGuest对象里直接获取即可。

2. 调整Celery配置与启动方式 #

优化ACK和预取设置

在你的Celery配置里补充以下参数：

CELERYD_PREFETCH_MULTIPLIER = 1  # 每次只预取1个任务，避免多任务堆积导致重复分配
CELERY_TASK_REJECT_ON_WORKER_LOST = True  # Worker意外退出时，任务回到队列（而非被标记为失败）
CELERY_TASK_MAX_RETRIES = 0  # 全局关闭默认重试，只在需要的任务里手动设置

分离Beat和Worker进程

你当前用--beat参数让Worker和Beat同进程运行，这可能导致调度逻辑出现意外冲突。建议分开启动：

• 启动Beat（负责调度任务）：

  celery beat --app openseat.tasks --loglevel info
  

• 启动Worker（负责执行任务）：

  celery worker --app openseat.tasks --concurrency 1 --loglevel info
  

3. 避免重复提交ETA任务 #

在调用apply_async时指定唯一的task_id，这样即使schedule_event_main被多次触发，Celery会忽略重复的任务ID：

# 生成唯一任务ID，结合event_id和任务类型
email_task_id = f"schedule_ratings_email_{event[0]}"
schedule_ratings_email.apply_async(
    args=[str(event[0])],
    eta=date_event,
    task_id=email_task_id
)

sms_task_id = f"schedule_ratings_sms_{event[0]}"
schedule_ratings_sms.apply_async(
    args=[str(event[0])],
    eta=date_event,
    task_id=sms_task_id
)

4. 设置任务超时时间 #

针对第三方服务的任务，设置合理的超时时间，防止任务长时间卡住导致Worker认为任务失败而重新调度：

# 给单个发送任务添加超时限制
@app.task(bind=True, soft_time_limit=30, hard_time_limit=60)
def schedule_individual_ratings_emails(self, guest_name, host, guest_email, url, guest_id, event_id):
    # 任务内容...

• soft_time_limit=30：30秒后抛出SoftTimeLimitExceeded异常，你可以捕获并处理
• hard_time_limit=60：60秒后强制终止任务

排查验证步骤 #

1. 给schedule_event_main添加日志，确认是否被重复触发
2. 用Redis命令查看ETA队列：KEYS celery*，检查celery_eta队列里是否有重复的任务ID
3. 查看Heroku dyno的重启日志，确认是否有频繁重启导致Worker重新加载任务

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