python - 定期任务在 Django Celery 中不起作用

Question

这是我的周期性任务:

from celery.task import PeriodicTask
from celery.registry import tasks
from datetime import timedelta
from datetime import datetime

class MyTask(PeriodicTask):
    run_every = timedelta(minutes=1)

    def run(self, **kwargs):
        self.get_logger().info("Time now: " + datetime.now())
        print("Time now: " + datetime.now())

tasks.register(MyTask)

我在 python manage.py runserver 启动了我的 django 服务器，还使用 ​​python manage.py celeryd -B 启动了 celery。

虽然根据代码，命令提示符上仍然没有打印任何内容，但时间应该已经打印出来了。

知道这里出了什么问题吗？看起来很简单。

Answer 1

你应该有这样的东西:

CELERYBEAT_SCHEDULE = {
    'proc': {
        "task": "tasks.processing",
        "schedule": timedelta(seconds=60),
        },
    }

在你的 celeryconfig.py 或在 django 的 settings.py在此处查看完整文档 http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html