python - significa - Cómo enviar tareas periódicas a una cola específica en Apio

qué significa en español

Question

De forma predeterminada, Celery envía todas las tareas a la cola ''apio'', pero puede cambiar este comportamiento agregando un parámetro extra:

@task(queue=''celery_periodic'')
def recalc_last_hour():
    log.debug(''sending new task'')
    recalc_hour.delay(datetime(2013, 1, 1, 2)) # for example

Configuración del programador:

CELERYBEAT_SCHEDULE = {
   ''installer_recalc_hour'': {
        ''task'': ''stats.installer.tasks.recalc_last_hour'',
        ''schedule'': 15  # every 15 sec for test
    },
}
CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"

Ejecutar trabajador:

python manage.py celery worker -c 1 -Q celery_periodic -B -E

Este esquema no funciona como se esperaba: estos trabajadores envían tareas periódicas a la cola de "apio", no a "apio_periodo". ¿Cómo puedo arreglar eso?

PS apio == 3.0.16

Answer 1

Encontré la solución para este problema:

1) En primer lugar, cambié la forma de configurar las tareas periódicas. Usé el decorador @periodic_task de esta manera:

@periodic_task(run_every=crontab(minute=''5''),
               queue=''celery_periodic'',
               options={''queue'': ''celery_periodic''})
def recalc_last_hour():
    dt = datetime.utcnow()
    prev_hour = datetime(dt.year, dt.month, dt.day, dt.hour) /
                - timedelta(hours=1)
    log.debug(''Generating task for hour %s'', str(prev_hour))
    recalc_hour.delay(prev_hour)

2) Escribí celery_periodic dos veces en params en @periodic_task :

• La opción queue = ''celery_periodic'' se usa cuando invocas una tarea desde el código (.delay o .apply_async)

• options = {''queue'': la opción ''celery_periodic''} se usa cuando Celery Beat lo invoca.

Estoy seguro de que lo mismo es posible si configurara tareas periódicas con la variable CELERYBEAT_SCHEDULE.

UPD. Esta solución es correcta tanto para el almacenamiento basado en DB como en el basado en archivos para CELERYBEAT_SCHEDULER .

Answer 2

Periódico se envían a las colas por celerybeat. Puedes hacer todo lo que hacemos con Celery api. Aquí está la lista de configuraciones que viene con celerybeat.

http://celery.readthedocs.org/en/latest/userguide/periodic-tasks.html#available-fields

En tu caso

CELERYBEAT_SCHEDULE = {
   ''installer_recalc_hour'': {
        ''task'': ''stats.installer.tasks.recalc_last_hour'',
        ''schedule'': 15  # every 15 sec for test,
        ''options'': {''queue'' : ''celery_periodic''} ##options are mapped to apply_async options
    },
}

Answer 3

Y si está utilizando el planificador de base de datos djcelery, puede especificar la cola en el campo Opciones de ejecución -> cola