Django和Celery的例子:周期性任务

Question

我一直在与Django/Celery文档争论一段时间,需要一些帮助.

我希望能够使用django-celery运行Periodic Tasks.我已经在互联网(和文档)周围看到了几种不同的格式和模式,以了解如何使用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.