芹菜周期性任务不执行 [英] celery periodic tasks not executing

问题描述

我正在学习芹菜，我创建了一个测试我的配置的项目。我根据最新的文件安装了 celery == 4.0.0 和 django-celery-beat == 1.0.1

在drf_project（带有manage.py的主项目目录）/drf_project/celery.py

  from __future__ import absolute_import，unicode_literals 
 from celery import Celery 
 import os 
 os.environ.setdefault（'DJANGO_SETTINGS_MODULE'，'drf_project.settings'）
 app = Celery（'drf_project'）
 app.config_from_object（'django.conf：settings'，namespace ='CELERY'）
 app.autodiscover_tasks（）
  

在drf_project / drf_project / settings.py

  INSTALLED_APPS + =（'django_celery_beat'，）
 CELERYBEAT_SCHEDULE = {
test_1：{
task：tasks.print_test，
schedule：timedelta秒= 2），
}，
} 
  

在drf_project / drf_project / init .py

  from __future __ import absolute_import，unicode_literals 
 from .celery import app as celery_app 
 
 __all__ = ['celery_app'] 
  

在我的user_management应用程序（drf_project / user_mangement /）中，我添加了一个tasks.py

 从芹菜进口Celery 
从时间import strftime 
 
 app = Celery（）
 
 @ app.task 
 def print_test（）：
 print strftime（'％Y-％m-％d％H：％M：％S'）
 with open（'abc.txt'，'ab +'）as test_file：
 test_file.writeline strftime（'％Y-％m-％d％H：％M：％S'））
  

当我运行芹菜工作者和我的django项目开发服务器在不同的终端通过：

 芹菜-A drf_project worker -l info 
  

和

  python manage.py runserver 
  

我可以在芹菜日志中看到我的任务： p>

  [tasks] 
。 user_management.tasks.print_test 
  

但它没有执行。我也没有得到任何错误。那我做错了什么？我遵循芹菜的正式文件。

解决方案

要运行定期任务，您必须运行芹菜 beat 而不是芹菜工作者。

您可以在本页尾部找到更多信息：
http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html

I am learning celery and I created a project to test my configuration. I installed celery==4.0.0 and django-celery-beat==1.0.1 according to the latest documentation.

In drf_project(main project dir with manage.py)/drf_project/celery.py

from __future__ import absolute_import, unicode_literals
from celery import Celery
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'drf_project.settings')
app = Celery('drf_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

In drf_project/drf_project/settings.py

INSTALLED_APPS += ('django_celery_beat',)
CELERYBEAT_SCHEDULE = {
    "test_1": {
        "task": "tasks.print_test",
        "schedule": timedelta(seconds=2),
    },
}

In drf_project/drf_project/init.py

from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app

__all__ = ['celery_app']

In my user_management app (drf_project/user_mangement/) I added a tasks.py

from celery import Celery
from time import strftime

app = Celery()

@app.task
def print_test():
    print strftime('%Y-%m-%d %H:%M:%S')
    with open('abc.txt', 'ab+') as test_file:
        test_file.writeline(strftime('%Y-%m-%d %H:%M:%S'))

when i run the celery worker and my django project dev server in different terminals by:

 celery -A drf_project worker -l info

and

 python manage.py runserver

I can see my task in celery log like:

[tasks]
   . user_management.tasks.print_test

But it is not executing. Also I am not getting any error. SO what I am doing wrong? I followed the official documentation of celery.

解决方案

For running periodic tasks you have to run celery beat instead of celery worker.

You can find more information at the end of this page: http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html

这篇关于芹菜周期性任务不执行的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！