Django芹菜中不工作的周期性任务

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 management .py runserver启动django服务器，也用python management .py celeryd -B启动芹菜。

仍然没有在命令提示符上打印，尽管根据代码，应该打印出时间。

你知道这里出了什么问题吗?

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

Answer 2

检查你的系统时区是否与Django的时区相同

对于Django:从settings文件中修改

对于linux操作系统:修改为:ln -s /usr/share/zoneinfo/America/Chicago /etc/localtime