celeryd and celerybeat - «Полученная незарегистрированная задача типа ...»

Question

ОК, ребята, это меня заводит. Я сейчас работаю над этим в ПОЛНЫЙ ДЕНЬ и не могу заставить его работать!
Моя структура сельдерей проекта заключается в следующем:

# celery.py 

from celery.schedules import crontab 
from celery import Celery 

celery = Celery('scheduler.celery', 
       include=['scheduler.tasks']) 

celery.config_from_object('celeryconfig') 

И:

# tasks.py 

from scheduler.celery import celery 

@celery.task 
def test(): 
    do_something() 

И:

# celeryconfig.py 

from celery.schedules import crontab 

CELERYBEAT_SCHEDULE = { 
    'test-cron': { 
     'task': 'tasks.test', 
     'schedule': crontab(minute='*/1'), 
    }, 
} 
# CELERY_IMPORTS = ('tasks',) 
BROKER_URL = 'redis://localhost:6379/0' 

Все файлы находятся в папке projects/scheduler/.
Когда я запускаю службу celeryd, я вижу, что она работает и подключается к моему брокеру, но когда я запускаю службу celerybeat, я могу видеть в журнале сообщение: Received unregistered task of type 'tasks.test'.
Если я раскомментирую константу CELERY_IMPORTS (как предлагается во многих ответах здесь, в SO), услуга celeryd даже не запустится! Фактически, он выводит OK, но с использованием ps ef | grep celery Я вижу, что он не работает.

Мой файл конф демон выглядит следующим образом:

# Name of nodes to start 
CELERYD_NODES="w1" 

# Where to chdir at start. 
CELERYD_CHDIR="/home/me/projects/scheduler/" 

# Extra arguments to celeryd 
CELERYD_OPTS="--time-limit=300 --concurrency=4" 

# %n will be replaced with the nodename. 
CELERYD_LOG_FILE="/var/log/celery/%n.log" 
CELERYD_PID_FILE="/var/run/celery/%n.pid" 

# Workers should run as an unprivileged user. 
CELERYD_USER="celery" 
CELERYD_GROUP="celery" 

# Extra arguments to celerybeat 
CELERYBEAT_OPTS="--schedule=/var/run/celerybeat-schedule" 

Любая помощь очень ценится.

Answer 1

Если вы не имеете scheduler модуль и запустить сельдерей из модуля корня:

runner.py:

from celery_test import celery 

if __name__ == '__main__': 
    celery.start() 

переименованной celery.py в celery_test.py:

from celery import Celery 

celery = Celery('scheduler.celery', 
       include=['tasks']) 

celery.config_from_object('celeryconfig') 

celeryconfig.py:

from celery.schedules import crontab 

CELERYBEAT_SCHEDULE = { 
    'test-cron': { 
     'task': 'tasks.test', 
     'schedule': crontab(minute='*/1'), 
    }, 
} 

BROKER_URL = 'redis://localhost:6379/0' 

tasks.py, фиксированный импорт:

from celery_test import celery 

@celery.task 
def test(): 
    do_something() 

Там вы должны быть осторожны с импортом и добавить runner.py, потому что импортировать celery_test, а затем tasks, внутри которого импортирования celery_test снова ,

---

Если у вас есть scheduler модуль и запустить сельдерей из корня модуля:

__init__.py пустой.

runner.py:

from scheduler.celery_test import celery 

if __name__ == '__main__': 
    celery.start() 

переименован celery.py в celery_test.py:

from celery import Celery 

celery = Celery('scheduler.celery', 
       include=['scheduler.tasks']) 

celery.config_from_object('celeryconfig')  

celeryconfig.py, установленное по умолчанию имя задачи:

from celery.schedules import crontab 

CELERYBEAT_SCHEDULE = { 
    'test-cron': { 
     'task': 'scheduler.tasks.test', 
     'schedule': crontab(minute='*/1'), 
    }, 
} 

BROKER_URL = 'redis://localhost:6379/0' 

tasks.py:

from scheduler.celery_test import celery 

@celery.task 
def test(): 
    do_something()