芹菜：访问上次运行任务的时间？

Posted 2023-02-23

【中文标题】芹菜：访问上次运行任务的时间？

【英文标题】：Celery: access to time of last run of task?

【发布时间】：2011-10-13 20:04:15

【问题描述】：

使用celery，是否可以在任务中检查上次运行该任务的时间是什么时候？

我想实现“抓取自上次运行以来的所有内容”。我可以自己跟踪上次运行的时间戳，也可以通过 celery 访问它。

这来自一个 django 项目，如果有影响的话。

【参考方案1】：

默认情况下，celery 不会永久存储任务结果，如果您在节拍模式下运行，则会运行一些干净的进程来获取干净的任务结果以及执行信息。 希望我建议您使用 NoSQL 存储每个最后执行日期，您可以通过在任务中覆盖方法 after_return 来做到这一点

def after_return(self, status, retval, task_id, args, kwargs, einfo):
    #exit point for context managers
    self.taskLogger.__exit__(status, retval, task_id, args, kwargs, einfo)

每次任务结束时都会调用此方法，无论结果如何，通过检查状态，您可以仅在任务以 SUCCESS 完成或实施最适合您需要的行为时存储您的日期。

【参考方案2】：

如果你使用 django-celery，last_run at 保存在模型中。使用普通的 django ORM 使用您的任务名称过滤 PeriodicTask 模型。

from djcelery.models import PeriodicTask

last_run = PeriodicTask.objects.only('last_run_at').get(task='TASK_NAME_HERE').last_run_at

【讨论】：

此方法依赖于知道数据库中的任务名称。有没有办法确定数据库中的哪个PeriodicTask 触发了您当前的任务运行？

django-celery 被认为是旧的集成，不再推荐github.com/celery/django-celery

【参考方案3】：

在最新版本的celery(5.1.2)中，可以通过这段代码获取任务的最后运行时间。

#tasks.py
from django_celery_beat.models import PeriodicTask

@shared_task(bind=True)
def backup_everyday(self)   
    try:
        last_run = PeriodicTask.objects.get(task=self.name).last_run_at
    except:
        last_run = None

每天备份只是一个示例功能。 PeriodicTask 模型有一个名为“last_run_at”的属性，它存储了任务的先前执行时间。