Django 和 Celery 的示例：定期任务

Question

我已经与 Django/Celery 文档作斗争一段时间了，需要一些帮助。

我希望能够使用 django-celery 运行定期任务。我在互联网（和文档）上看到了几种不同的格式和模式，关于如何使用 Celery 来实现这一目标...

有人可以帮助提供一个 django-celery 的创建、注册和执行的基本功能示例吗？周期性任务？特别是，我想知道是否应该编写一个扩展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():
    ....

装饰器语法简单地允许您可以将现有功能/任务转变为周期性任务，而无需直接修改它们。

对于要执行的任务 celerybeat 必须正在运行< /a>.