检查Celery中计划的定期任务的下一个运行时间(使用Django)

2023年7月12日 255次阅读

*使用芹菜3.1.25因为
django-celery-beat 1.0.1存在调度周期性任务的问题.

最近我遇到了celerybeat的问题,其中一天或更长时间间隔的周期性任务似乎被调度程序“忘记”了.如果我将间隔更改为每5秒,则任务正常执行(每5秒)并且last_run_at属性会更新.这意味着celerybeat在一定程度上响应调度程序,但是如果我重置last_run_at,即PeriodicTask.objects.update(last_run_at = None),则不会再运行每天间隔的任务.

Celerybeat崩溃了一点,可能已经损坏了一些东西,所以我创建了一个新的virtualenv和数据库,看看问题是否仍然存在.我想知道是否有办法检索下一次运行时间,这样我就不必等一天知道我的周期性任务是否已经执行.

我也尝试过使用inspect< active / scheduled / reserved>但所有人都空了.对于使用djcelery的数据库调度程序的周期性任务,这是正常的吗？

这是调度任务的函数：

def schedule_data_collection(request, project):
    if (request.method == 'POST'):
        interval = request.POST.get('interval')

        target_project = Project.objects.get(url_path=project)    
        interval_schedule = dict(every=json.loads(interval), period='days')

        schedule, created = IntervalSchedule.objects.get_or_create(
            every=interval_schedule['every'],
            period=interval_schedule['period'],
        )

        task_name = '{} data collection'.format(target_project.name)

        try:
            task = PeriodicTask.objects.get(name=task_name)
        except PeriodicTask.DoesNotExist:
            task = PeriodicTask.objects.create(
                interval=schedule,
                name=task_name,
                task='myapp.tasks.collect_tool_data',
                args=json.dumps([target_project.url_path])
            )
        else:
            if task.interval != schedule:
                task.interval = schedule

            if task.enabled is False:
                task.enabled = True

            task.save()

        return HttpResponse(task.interval)
    else:
        return HttpResponseForbidden()

最佳答案 您可以通过进入shell并查看app.conf.CELERYBEAT_SCEDULE来查看调度程序.

celery -A myApp shell
print(app.conf.CELERYBEAT_SCHEDULE)

这应该显示您的所有定期任务.

点赞