Периодические задания сельдерея не выполняются

Question

Я изучаю сельдерей, и я создал проект для проверки моей конфигурации. Я установил 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(seconds=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_manageme нт приложение (drf_project/user_mangement /) Я добавил 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')) 

, когда я бегу сельдерея рабочий и мой Джанго проект Dev сервера в разных терминалах по:

celery -A drf_project worker -l info 

и

python manage.py runserver 

Я вижу свою задачу в журнале сельдерея, как:

[tasks] 
    . user_management.tasks.print_test 

Но он не выполняется. Также я не получаю никаких ошибок. ТАК, что я делаю неправильно? Я следовал официальной документации сельдерея.

Answer 1

Для выполнения периодических задач вам необходимо запустить сельдерей beat вместо работника сельдерея.

Вы можете найти более подробную информацию в конце этой страницы: http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html