python之如何使用 celery 在 Django 上设置周期性任务

我当前的代码:

from celery.task.schedules import crontab 
from celery.decorators import task, periodic_task 
 
 
@periodic_task(run_every=crontab(hour=15, minute=55, day_of_week="wed")) 
def demo(): 
    print("testing------------------------") 

设置文件:

CELERY_BROKER_URL = os.environ.get('CELERY_BROKER_URL', 'amqp://guest:guest@localhost//') 

它不起作用，我是不是漏掉了什么？

提前致谢

请您参考如下方法:

我认为你应该使用下面的代码，它应该适合你。

celery_app = celery("project_name") 
 
@celery_app.on_after_finalize.connect 
def setup_periodic_tasks(sender, **kwargs): 
  sender.add_periodic_task(5.0, demo.s(args)) 
 
@celery_app.task(bind=True) 
def demo(args): 
   print("testing------------------------")