Задача планирования сельдерея периодически, но не выполняется.

Question

У моего приложения django есть две задачи, из которых одна является периодической задачей.

нормальная задача: AddScore периодическая задача: CalculateTopScore

class CalculateTopScore(celery.Task): 

    default_retry_delay = settings.DEFAULT_RETRY_DELAY 
    max_retries = settings.DEFAULT_MAX_RETRIES 
    name = 'games.tasks.CalculateTopScore' 

    def run(self): 
     try: 
      # Code to run 

     except Exception, err: 
      logger.exception("Error in running task calculate_top_score") 
      self.retry(exc=err) 

     return True 

    def on_failure(self, exc, task_id): 

     failure = "%s task for calculate_top_score failed permanently." % task_id 
     logger.error(failure) 

    def on_success(self): 

     task_info = 'calculate_top_score task successfully' 
     logger.info(task_info) 

Я хочу, чтобы выполнить эту задачу периодически через каждые 30 минут.

Вот настройки, которые я использую:

#celery/settings.py 

import djcelery 
import kombu 

from celery.schedules import crontab 
from config.celery import exchanges 


djcelery.setup_loader() 

CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler" 

CELERYBEAT_SCHEDULE = { 
    "calcluate_score": { 
     "task": "games.tasks.CalculateTopScore", 
     "schedule": crontab(minute='*/30'), 
     "args":(), 
    }, 
} 

CELERY_QUEUES = (
    kombu.Queue('add_score', 
       exchange=exchanges.add_score_exchange, 
       routing_key='add.scores'), 
    kombu.Queue('calcluate_score', 
       exchange=exchanges.calcluate_score_exchange, 
       routing_key='calculate.scores'), 
) 

CELERY_ROUTES = ('config.celery.routers.CeleryTaskRouter',) 

# Default delay(in seconds) for retrying tasks. 
DEFAULT_RETRY_DELAY = 60 

# Maximum retry count 
DEFAULT_MAX_RETRIES = 6 

CELERY_IGNORE_RESULT = True 

exchanges.py файл

#exchanges.py 

from kombu import Exchange 

add_score_exchange = Exchange('add_score', type='direct') 
calcluate_score_exchange = Exchange('calcluate_score', type='direct') 

routes.py файл

ROUTES = { 
    'players.tasks.AddScore': { 
     'exchange': 'add_score', 
     'exchange_type': 'direct', 
     'routing_key': 'add.score', 
    }, 
    'games.tasks.CalculateTopScore': { 
     'exchange': 'calculate_score', 
     'exchange_type': 'direct', 
     'routing_key': 'calculate.score', 
    }, 
} 


class CeleryTaskRouter(object): 
    """ This is a basic celery task router. 
    """ 

    def route_for_task(self, task, arg=None, kwargs=None): 
     return ROUTES.get(task) 

На нашем производственном сервере я запустить celeryd с следующие аргументы: celeryd worker -B

Теперь в журналах я замечаю, что celerybeat расписывает задачу каждые 30 минут, но работник вообще не знает о запланированной задаче и, следовательно, ее не выполняет.

Почему? все конфигурации/настройки отсутствуют? Как периодически выполнять задачу на основе класса?

Пожалуйста, помогите

Answer 1

Вы, вероятно, придется указать очереди, которые вы хотите сельдерея, чтобы получить задачи из. По умолчанию используется очередь celery.

Try:

celeryd worker -B -Q add_score,calculate_score