Celery beat неправильно порождает периодические задания, несколько раз заполняя очередь - PullRequest
Новая
       34

Celery beat неправильно порождает периодические задания, несколько раз заполняя очередь

0 голосов
/ 03 ноября 2019

Я разрабатываю приложение Django с диспетчером задач сельдерея для выполнения 6 периодических задач со следующим расписанием: 

{
    'daily_full_import_provider1': {
        'task': 'update_data.tasks.import_full_provider_data',
        'schedule': '<crontab: * */2 * * * (m/h/d/dM/MY)>',
        'kwargs': {
            'provider': 'provider1'
        }
    },
    'daily_full_import_provider2': {
        'task': 'update_data.tasks.import_full_provider_data',
        'schedule': '<crontab: * */2 * * * (m/h/d/dM/MY)>',
        'kwargs': {
            'provider': 'provider2'
        }
    },
    'daily_full_import_provider3': {
        'task': 'update_data.tasks.import_full_provider_data',
        'schedule': '<crontab: * */2 * * * (m/h/d/dM/MY)>',
        'kwargs': {
            'provider': 'provider3'
        }
    },
    'daily_full_import_provider4': {
        'task': 'update_data.tasks.import_full_provider_data',
        'schedule': '<crontab: * */2 * * * (m/h/d/dM/MY)>',
        'kwargs': {
            'provider': 'provider4'
        }
    },
    'daily_full_import_provider5': {
        'task': 'update_data.tasks.import_full_provider_data',
        'schedule': '<crontab: * */2 * * * (m/h/d/dM/MY)>',
        'kwargs': {
            'provider': 'provider5'
        }
    },
    'daily_full_import_provider6': {
        'task': 'update_data.tasks.import_full_provider_data',
        'schedule': '<crontab: * */2 * * * (m/h/d/dM/MY)>',
        'kwargs': {
            'provider': 'provider6'
        }
    }
}

Каждая задача вызывает одну и ту же функцию python с различными параметрами: 

def provider_shadow_name(task, args, kwargs, options):
    return "celery.import.{}".format(kwargs['provider'])


@celery.task(shadow_name=provider_shadow_name)
def import_full_provider_data(provider=None):
    return ImportWorker(
        ProviderService.UPDATE_OP,
        provider=[provider] if provider else provider
    )()

Я запускаю один экземпляр worker / beat / flower на экране: 

screen -S djapp_uat_celery_worker -d -m -h 500 \
    $djapp_ENV/bin/celery -E -A djapp worker \
    --concurrency=$djapp_CELERY_THREAD_WORKERS \
    --pidfile=$djapp_VAR/djapp-celeryworker.pid \
    --loglevel=INFO \
    --logfile=$djapp_LOG_DIR/%p.%i.celery.worker.log

screen -S djapp_uat_celery_scheduler -d -m -h 500 \
    $djapp_ENV/bin/celery -A djapp beat \
    --schedule=$djapp_VAR/djapp-celerybeat-schedule.db \
    --pidfile=$djapp_VAR/djapp-celerybeat.pid \
    --loglevel=INFO \
    --logfile=$djapp_LOG_DIR/celery.scheduler.log

screen -S djapp_uat_celery_flower -d -m -h 500 $djapp_ENV/bin/celery -A djapp flower

Работник может одновременно выполнять до 3 задач. Все 6 заданий не должны занимать более 30 минут.

Я использую redis в качестве посредника сообщений со следующей конфигурацией: 

celery:
  celery_broker_url: redis://docker.local:6379
  celery_result_backend: django-db
  worker_redirect_stdouts_level: INFO
  worker_hijack_root_logger: false

Проблема в том, что через несколько часов показ цветов заканчивается400 (!) Сообщений в очереди брокера. Рабочий пытается очистить очередь, и он все время выбирает задачи. Тем не менее, beat ничего не показывает в выводе: 

celery beat v4.3.0 (rhubarb) is starting.
__    -    ... __   -        _
LocalTime -> 2019-11-03 11:42:13
Configuration ->
    . broker -> redis://localhost:6379//
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> celery.beat.PersistentScheduler
    . db -> /home/QA/uat/djapp/var/djapp-celerybeat-schedule.db
    . logfile -> /home/QA/uat/djapp/var/log/celery.scheduler.log@%INFO
    . maxinterval -> 5.00 minutes (300s)

Приложение Beat должно информировать stdout об отправке запланированных задач, но ничего не показывает ..

У меня заканчиваются идеи, какотладьте это и выясните причину ..

...