Django和芹菜的例子:周期性任务

Question

我已经和Django/芹菜文档斗争了一段时间了，需要一些帮助。

我希望能够使用django-芹菜运行周期性任务。我在互联网(和文档)上看到了几种不同的格式和模式，如何使用芹菜来实现这一点…

有人可以帮助一个基本的，功能的例子，创建，注册和执行django-芹菜周期任务?特别是，我想知道我是否应该编写一个扩展PeriodicTask类并注册它的任务，或者我是否应该使用@periodic_task装饰器，或者我是否应该使用@task装饰器然后为任务的执行设置一个时间表。

如果这三种方法都可行，我不介意，但我希望看到至少一种方法有效的例子。非常感谢您的帮助。

Answer 1

文档中的示例有什么问题?

from celery.task import PeriodicTask
from clickmuncher.messaging import process_clicks
from datetime import timedelta

class ProcessClicksTask(PeriodicTask):
    run_every = timedelta(minutes=30)
    def run(self, **kwargs):
        process_clicks()

你可以用一个装饰符来写同样的任务:

from celery.task.schedules import crontab
from celery.task import periodic_task
@periodic_task(run_every=crontab(minute="*/30"))
def process_clicks():
    ....

decorator语法允许您将现有的函数/任务转换为周期性任务，而无需直接修改它们。

对于要执行的任务，celerybeat必须正在运行