Сложности для настройки простой задачи

Question

Я пытаюсь настроить простую задачу (печать "foo" каждые 5 секунд) с помощью django и celery.

Предполагая, что у меня есть следующий проект:

. 
├── manage.py 
└── proj 
    ├── __init__.py 
    ├── settings.py 
    ├── tasks.py 
    ├── urls.py 
    └── wsgi.py 

с tasks.py:

from celery import Celery 
import os 

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') 

from django.conf import settings 

app = Celery('tasks', broker='django://127.0.0.1:8000//') 

app.config_from_object('django.conf:settings') 
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) 

@app.task 
def test(): 
    print("FOO") 

И settings.py:

INSTALLED_APPS = [ 
    'django.contrib.admin', 
    'django.contrib.auth', 
    'django.contrib.contenttypes', 
    'django.contrib.sessions', 
    'django.contrib.messages', 
    'django.contrib.staticfiles', 
    'celery', 
    'kombu.transport.django', 
] 

[...] 

# Celery :                         

CELERYBEAT_SCHEDULE = { 
    'test-celery': { 
    'task' : 'proj.tasks.test', 
     'schedule' : timedelta(seconds=5) 
    }, 
} 

CELERY_TIMEZONE = 'Europe/Paris' 

и proj/__init__.py:

from .tasks import app as celery_app 

Я мигрируют, запуск с celery -A proj.tasks worker --loglevel=info и сельдерея запущен, я просто два предупреждения с использованием рассола с этой версии (3.1.23 ) и о settings.DEBUG быть True. Но сельдерей кажется работающим. Я также запускаю свой dev-сервер с manage.py.

Так почему же не моя задача (test()) работает?

Answer 1

Вам необходимо запустить Celery Beat. Например, например:

celery -A proj worker -l info -B