python之Django Celery 周期性任务示例

我需要一个最小示例来执行周期性任务(每 5 分钟后运行一些函数，或者在 12:00:00 运行一些函数等)。

在我的 myapp/tasks.py 中，我有，

from celery.task.schedules import crontab 
from celery.decorators import periodic_task 
from celery import task 
 
 
@periodic_task(run_every=(crontab(hour="*", minute=1)), name="run_every_1_minutes", ignore_result=True) 
def return_5(): 
    return 5 
 
 
@task 
def test(): 
    return "test" 

当我运行 celery workers 时，它会显示任务(如下所示)但不返回任何值(在终端或花中)。

[tasks] 
  . mathematica.core.tasks.test 
  . run_every_1_minutes 

请提供实现预期结果的最少示例或提示。

背景:

我有一个 config/celery.py，其中包含以下内容:

import os 
from celery import Celery 
 
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.local") 
 
app = Celery('config') 
app.config_from_object('django.conf:settings', namespace='CELERY') 
app.autodiscover_tasks() 

在我的 config/__init__.py 中，我有

from .celery import app as celery_app 
 
__all__ = ['celery_app'] 

我在 myapp/tasks.py

中添加了类似下面的函数

from celery import task 
 
@task 
def test(): 
    return "test" 

当我从 shell 运行 test.delay() 时，它运行成功，并且在 flower 中显示了任务信息

请您参考如下方法:

要运行周期性任务，你应该运行 celery beat还。您可以使用此命令运行它:

celery -A proj beat 

或者，如果您使用一名 worker :

celery -A proj worker -B