Celery Worker异常重启后禁止执行过期历史定时任务方案咨询

你之前的自定义Task基类方案无效的原因是：before_start钩子触发时，Worker已经处于运行状态，inspect.active()必然能检测到当前Worker，导致判断逻辑完全失效。以下是两种可行的解决方案：

方案一：给任务设置过期时间（最简单） #

直接给定时任务添加expires参数，指定任务的有效时长。超过该时长后，Broker会自动丢弃任务，Worker重启后不会再处理这些过期任务。

如果你的crontab任务是每分钟执行一次，可设置expires=60（单位：秒）：

from celery import Task, Celery

app = Celery('tasks', broker='your_broker_url')

class MyTask(Task):
    pass

@app.task(base=MyTask, expires=60)
def test():
    return 'test'

也可以在自定义基类中统一设置，避免每个任务重复配置：

class MyTask(Task):
    expires = 60  # 所有继承该基类的任务默认60秒后过期

@app.task(base=MyTask)
def test():
    return 'test'

方案二：自定义任务执行前的过期判断（更灵活） #

通过任务的on_prerun钩子，对比当前时间与任务的预期执行时间（self.request.eta），如果延迟超过设定阈值，直接终止任务执行。这种方式适合需要自定义过期规则的场景。

from datetime import datetime, timezone
from celery import Task, Celery

app = Celery('tasks', broker='your_broker_url')

class MyTask(Task):
    def on_prerun(self, task_id, args, kwargs):
        eta = self.request.eta
        if eta:
            # Celery默认用UTC存储时间，需统一时区对比
            now = datetime.now(timezone.utc)
            # 设置允许的最大延迟时间，比如5分钟（300秒）
            max_allowed_delay = 300
            if (now - eta).total_seconds() > max_allowed_delay:
                print(f"任务 {task_id} 已过期，跳过执行")
                # 拒绝任务且不重新入队，直接终止
                raise self.reject(exc=Exception("任务已过期"), requeue=False)
        return super().on_prerun(task_id, args, kwargs)

@app.task(base=MyTask)
def test():
    return 'test'

注意事项 #

• 时区问题：务必确保当前时间与eta的时区一致（优先用UTC），否则会出现判断误差。
• Broker支持：expires参数需要你的Broker（如Redis、RabbitMQ）支持任务过期清理，大部分主流Broker都兼容。

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